/*
Copyright (C) 2000-2010  Ministere de la culture et de la communication (France), AJLSM
See LICENCE file
*/
package fr.gouv.culture.sdx.oai;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.TreeMap;
import java.util.Vector;

import org.apache.avalon.framework.configuration.Configuration;
import org.apache.avalon.framework.configuration.ConfigurationException;
import org.apache.avalon.framework.context.ContextException;
import org.apache.avalon.framework.parameters.Parameters;
import org.apache.avalon.framework.service.ServiceException;
import org.apache.cocoon.xml.IncludeXMLConsumer;
import org.apache.cocoon.xml.XMLPipe;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.index.Term;
import org.apache.lucene.search.BooleanClause;
import org.apache.lucene.search.BooleanQuery;
import org.apache.lucene.search.Hits;
import org.apache.lucene.search.TermQuery;
import org.apache.lucene.search.WildcardQuery;
import org.xml.sax.SAXException;
import org.xml.sax.helpers.AttributesImpl;

import fr.gouv.culture.oai.OAIError;
import fr.gouv.culture.oai.OAIObject;
import fr.gouv.culture.oai.OAIRequest;
import fr.gouv.culture.oai.util.OAIUtilities;
import fr.gouv.culture.sdx.document.XMLDocument;
import fr.gouv.culture.sdx.documentbase.LuceneDocumentBase;
import fr.gouv.culture.sdx.exception.SDXException;
import fr.gouv.culture.sdx.exception.SDXExceptionCode;
import fr.gouv.culture.sdx.framework.Framework;
import fr.gouv.culture.sdx.pipeline.Pipeline;
import fr.gouv.culture.sdx.search.lucene.DateField;
import fr.gouv.culture.sdx.search.lucene.FieldList;
import fr.gouv.culture.sdx.search.lucene.analysis.MetaAnalyzer;
import fr.gouv.culture.sdx.search.lucene.query.DateIntervalQuery;
import fr.gouv.culture.sdx.search.lucene.query.LuceneIndex;
import fr.gouv.culture.sdx.search.lucene.query.SearchLocations;
import fr.gouv.culture.sdx.search.lucene.query.SimpleQuery;
import fr.gouv.culture.sdx.search.lucene.query.Terms;
import fr.gouv.culture.sdx.utils.Utilities;
import fr.gouv.culture.sdx.utils.configuration.ConfigurationUtils;
import fr.gouv.culture.sdx.utils.constants.ContextKeys;
import fr.gouv.culture.sdx.utils.database.DatabaseEntity;

//TODO: better abstraction here

public class LuceneDocumentBaseOAIRepository extends AbstractDocumentBaseOAIRepository {


    /**Search index of the underlying LuceneDocumentBase*/
    protected LuceneIndex luceneSearchIndex = null;
    
    /**query string for document selection*/
    protected SimpleQuery includeQuery = null;
    /**query string for document selection*/
    protected SimpleQuery excludeQuery = null;

    /**Set mappings from configuration file*/
    protected Hashtable setMappings = null;
    
    protected final String ATTRIBUTE_NAME_SDXFIELD = "sdxField";


    protected final String ELEMENT_NAME_OAI_FORMAT = "oai-format";
    
    /*ResumptionToken in request*/
    protected String resumptionToken;
    /*ResumptionToken in response*/
    protected String newResumptionToken;

    /*Lucene Specific*/
    protected final String ELEMENT_NAME_OAI_SUBSET = "oai-subset";
    protected final String ELEMENT_NAME_INCLUDE = "include";
    protected final String ELEMENT_NAME_EXCLUDE = "exclude";
    protected static final String ATTRIBUTE_NAME_INCLUDE_QUERY = "includeQuery";
    protected static final String ATTRIBUTE_NAME_EXCLUDE_QUERY = "excludeQuery";
    public static final String PARAMETER_NAME_INCLUDE_QUERY = ATTRIBUTE_NAME_INCLUDE_QUERY;
    public static final String PARAMETER_NAME_EXCLUDE_QUERY = ATTRIBUTE_NAME_EXCLUDE_QUERY;


    public LuceneDocumentBaseOAIRepository(LuceneDocumentBase base) {
        if (base != null) {
            this.documentBase = base;
            this.documentBaseId = base.getId();
            this.luceneSearchIndex = (LuceneIndex) base.getIndex();
        }
        //indicating the protocol version supported
        super.protocolVersion = "2.0";
        //yyyy-MM-dd'T'HH:mm:ssz this is the java date format string we use in lucene document base
        //A repository that supports YYYY-MM-DDThh:mm:ssZ should indicate so in the Identify response
        super.granularity = "YYYY-MM-DDThh:mm:ssZ";
        //indicating level of deleted record support according to OAI-PMH guidelines
        super.deletedRecord = "transient";
    }


    /**Configure an LuceneDocumentBaseOAIRepository
     * @param confugiration	The configuration
     */
    public void configure(Configuration configuration) throws ConfigurationException {
        //should configure protocolVersion, granularity, deleteRecord, repositoryName, and admin email
        super.configure(configuration);
        configureSets(configuration);
        configureFormats(configuration);
        configureSubset(configuration);
    }

    /**Configure the sets of an OAIRepository
     * @param confugiration	The configuration
     */
    protected void configureSets(Configuration configuration) throws ConfigurationException {
        Configuration setsConf = configuration.getChild(DocumentBaseOAIRepository.ConfigurationNode.OAI_SETS, false);
        if (setsConf != null) {
            Configuration[] sets = setsConf.getChildren(DocumentBaseOAIRepository.ConfigurationNode.OAI_SET);
            for (int i = 0; i < sets.length; i++) {
                Configuration set = sets[i];
                if (set != null) {
                    String sdxFieldName = set.getAttribute(ATTRIBUTE_NAME_SDXFIELD, "");
                    String setName = set.getAttribute(DocumentBaseOAIRepository.ConfigurationNode.NAME, null);
                    String setSpec = set.getAttribute(DocumentBaseOAIRepository.ConfigurationNode.SPEC, sdxFieldName);//using sdxFieldName as default setSpec
                    String includeQuery = set.getAttribute(ATTRIBUTE_NAME_INCLUDE_QUERY, "");
                    String excludeQuery = set.getAttribute(ATTRIBUTE_NAME_EXCLUDE_QUERY, "");
                    ConfigurationUtils.checkConfAttributeValue(ATTRIBUTE_NAME_SDXFIELD, sdxFieldName, set.getLocation());
                    ConfigurationUtils.checkConfAttributeValue(DocumentBaseOAIRepository.ConfigurationNode.NAME, setName, set.getLocation());
                    if (this.luceneSearchIndex != null) {
                        MetaAnalyzer ma = this.luceneSearchIndex.getMetaAnalyzer();
                        if (ma != null) {
                            FieldList fieldsDef = ma.getFieldList();
                            if (fieldsDef != null) {
                                fr.gouv.culture.sdx.search.lucene.Field sdxField = fieldsDef.getField(sdxFieldName);
                                if (sdxField != null && !sdxField.isInBrief()) {
                                    String[] args = {sdxFieldName};
                                    SDXException sdxE = new SDXException(logger, SDXExceptionCode.ERROR_SDX_OAI_FIELD_NOT_BRIEF, args, null);
                                    throw new ConfigurationException(sdxE.getMessage(), sdxE);
                                }
                            }
                        }
                    }
                    Parameters setParams = new Parameters();
                    if (Utilities.checkString(setName))
                        setParams.setParameter(PARAMETER_NAME_SET_NAME, setName);
                    if (Utilities.checkString(sdxFieldName))
                        setParams.setParameter(PARAMETER_NAME_SDX_FIELD, sdxFieldName);
                    if (Utilities.checkString(setSpec))
                        setParams.setParameter(PARAMETER_NAME_SET_SPEC, sdxFieldName);
                    try {
                        if (Utilities.checkString(includeQuery)) {
                            getLuceneSimpleQuery(includeQuery);//parsing the query to reveal any config problems
                            setParams.setParameter(PARAMETER_NAME_INCLUDE_QUERY, includeQuery);
                        }

                        if (Utilities.checkString(excludeQuery)) {
                            getLuceneSimpleQuery(excludeQuery);//parsing the query to reveal any config problems
                            setParams.setParameter(PARAMETER_NAME_EXCLUDE_QUERY, excludeQuery);
                        }
                        if (this.setMappings == null) this.setMappings = new Hashtable();
                        Utilities.isObjectUnique(this.setMappings, setSpec, setParams);
                        this.setMappings.put(setSpec, setParams);
                    } catch (SDXException e) {
                        throw new ConfigurationException(e.getMessage(), e);
                    }
                }
            }

        }
    }


    /**Configure the supported formats of an OAIRepository
     * @param confugiration	The configuration
     */
    protected void configureFormats(Configuration configuration) throws ConfigurationException {
        //should configure metadata formats here from config file
        Configuration[] formats = configuration.getChildren(ELEMENT_NAME_OAI_FORMAT);
        if (formats.length > 0) super.metadataFormats = new Hashtable();
        for (int i = 0; i < formats.length; i++) {
            BasicOAIMetadataFormat format = null;
            String formatPrefix = formats[i].getAttribute(OAIObject.Node.Name.METADATA_PREFIX);
            if (formatPrefix.equalsIgnoreCase(Framework.SDXNamespacePrefix)) {
                format = new SDXOAIMetadataFormat();
            } else
                format = new BasicOAIMetadataFormat();
            try {
                format.enableLogging(this.logger);
                format.contextualize(Utilities.createNewReadOnlyContext(getContext()));
                format.service(this.manager);
                format.configure(formats[i]);
                formatPrefix = format.getPrefix();
                if (Utilities.checkString(formatPrefix)) {
                    Utilities.isObjectUnique(super.metadataFormats, formatPrefix, format);
                    super.metadataFormats.put(formatPrefix, format);
                }
            } catch (ServiceException e) {
                throw new ConfigurationException(e.getMessage(), e);
            } catch (SDXException e) {
                throw new ConfigurationException(e.getMessage(), e);
            } catch (ContextException e) {
                throw new ConfigurationException(e.getMessage(), e);
            }

        }

        if (!super.metadataFormats.containsKey(OAIObject.Node.Prefix.OAI_DC)) {
            String[] args = new String[1];
            args[0] = this.documentBaseId;
            SDXException sdxE = new SDXException(logger, SDXExceptionCode.ERROR_NO_OAI_DC_METADATA_FORMAT, args, null);
            throw new ConfigurationException(sdxE.getMessage(), sdxE);
        }
    }


    protected void configureSubset(Configuration configuration) throws ConfigurationException {
        //the subset of documents which should be include and excluded via lucene query syntax
        Configuration oaiSubset = configuration.getChild(ELEMENT_NAME_OAI_SUBSET, false);
        if (oaiSubset != null) {

            SearchLocations sLocs = this.getSearchLocation();

            Configuration include = oaiSubset.getChild(ELEMENT_NAME_INCLUDE, false);
            Configuration exclude = oaiSubset.getChild(ELEMENT_NAME_EXCLUDE, false);
            
            /*include and no exclude*/
            if (include != null && exclude == null) {
                String includeQueryString = include.getAttribute(DocumentBaseOAIRepository.ConfigurationNode.QUERY);
                if (Utilities.checkString(includeQueryString)) {

                    this.includeQuery = new SimpleQuery();
                    this.includeQuery.enableLogging(this.logger);
                    try {
                        this.includeQuery.setUp(sLocs, null, includeQueryString);
                    } catch (SDXException e) {
                        throw new ConfigurationException(e.getMessage(), e);
                    }
                }

            }

            /*exclude and no include*/
            if (exclude != null && include == null) {
                String excludeQueryString = exclude.getAttribute(DocumentBaseOAIRepository.ConfigurationNode.QUERY);
                if (Utilities.checkString(excludeQueryString)) {
                    this.excludeQuery = new SimpleQuery();
                    this.excludeQuery.enableLogging(this.logger);
                    try {
                        this.excludeQuery.setUp(sLocs, null, excludeQueryString);
                    } catch (SDXException e) {
                        throw new ConfigurationException(e.getMessage(), e);
                    }
                }

            }
            
            /*both include and exclude*/
            if (exclude != null && include != null) {
            	String excludeQueryString = exclude.getAttribute(DocumentBaseOAIRepository.ConfigurationNode.QUERY);
                if (Utilities.checkString(excludeQueryString)) {
                    this.excludeQuery = new SimpleQuery();
                    this.excludeQuery.enableLogging(this.logger);
                    try {
                        this.excludeQuery.setUp(sLocs, null, excludeQueryString);
                    } catch (SDXException e) {
                        throw new ConfigurationException(e.getMessage(), e);
                    }
                }
            	String includeQueryString = include.getAttribute(DocumentBaseOAIRepository.ConfigurationNode.QUERY);
                if (Utilities.checkString(includeQueryString)) {

                    this.includeQuery = new SimpleQuery();
                    this.includeQuery.enableLogging(this.logger);
                    try {
                        this.includeQuery.setUp(sLocs, null, includeQueryString);
                    } catch (SDXException e) {
                        throw new ConfigurationException(e.getMessage(), e);
                    }
                }
            }
            
            OAIUtilities.logDebug(logger, "Configured AOI subset: include: " + 
            		((this.includeQuery==null)?"":this.includeQuery.getLuceneQuery().toString()) +
            		((this.excludeQuery==null)?"":this.excludeQuery.getLuceneQuery().toString())
            		);

        }

    }


    public void listSets(OAIRequest request) throws SAXException {
        /*
        Summary and Usage Notes
       This verb is used to retrieve the set structure of a repository, useful for selective harvesting.

       Arguments
       resumptionToken an exclusive argument with a value that is the flow control token returned by a previous ListSets request that issued an incomplete list
       Error and Exception Conditions
       badArgument - The request includes illegal arguments or is missing required arguments.
       badResumptionToken - The value of the resumptionToken argument is invalid or expired.
       noSetHierarchy - The repository does not support sets.
       */
        if (this.setMappings == null || this.setMappings.size() == 0)
            super.sendNoSetHierarchyError();
        else if (Utilities.checkString(request.getResumptionToken())) {
            new OAIError(OAIError.ERROR_BAD_RESUMPTION_TOKEN, "This repository does not support resumptionTokens for ListSets requests").toSAX(this);
        } else {
            super.startVerbEvent(request);
            Enumeration sets = this.setMappings.keys();
            if (sets != null) {
                while (sets.hasMoreElements()) {
                    String setSpec = (String) sets.nextElement();
                    if (Utilities.checkString(setSpec)) {

                        Parameters setParams = (Parameters) this.setMappings.get(setSpec);
                        String setName = "";
                        String sdxFieldName = "";
                        if (setParams != null) {
                            setName = setParams.getParameter(PARAMETER_NAME_SET_NAME, "");
                            sdxFieldName = setParams.getParameter(PARAMETER_NAME_SDX_FIELD, "");
                        }
                        Terms terms = new Terms();
                        if (Utilities.checkString(sdxFieldName)) {
                            terms.enableLogging(this.logger);
                            try {
                                terms.setUp(getSearchLocation(), sdxFieldName, "*");
                            } catch (SDXException e) {
                            	OAIUtilities.logException(logger, e);
                            }
                        }

                        TreeMap termsMap = terms.getList();
                        if (termsMap != null && termsMap.size() > 0) {
                            //using each field value for the defined set
                            Iterator termsItr = termsMap.keySet().iterator();
                            while (termsItr.hasNext()) {
                                String key = (String) termsItr.next();
                                setSpec = key;

                                super.startElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.SET, OAIObject.Node.Name.SET, null);

                                if (Utilities.checkString(setName))
                                    super.sendElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.SET_NAME, OAIObject.Node.Name.SET_NAME, null, setName);
                                if (Utilities.checkString(setSpec))
                                    super.sendElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.SET_SPEC, OAIObject.Node.Name.SET_SPEC, null, setSpec);

                                super.endElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.SET, OAIObject.Node.Name.SET);


                            }
                        } else {
                            //using the application.xconf definitions
                            super.startElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.SET, OAIObject.Node.Name.SET, null);

                            if (Utilities.checkString(setName))
                                super.sendElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.SET_NAME, OAIObject.Node.Name.SET_NAME, null, setName);
                            if (Utilities.checkString(setSpec))
                                super.sendElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.SET_SPEC, OAIObject.Node.Name.SET_SPEC, null, setSpec);

                            super.endElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.SET, OAIObject.Node.Name.SET);
                        }
                    }
                }
            }

            super.endVerbEvent(request);
        }

    }


    public void listIdentifiers(OAIRequest request) throws SAXException {
        /*
        Summary and Usage Notes
        This verb is an abbreviated form of ListRecords, retrieving only headers rather than records. Optional arguments permit selective harvesting of headers based on set membership and/or _datestamp. Depending on the repository's support for deletions, a returned header may have a status attribute of "deleted" if a record matching the arguments specified in the request has been deleted.

        Arguments
        from an optional argument with a UTCdatetime value , which specifies a lower bound for _datestamp-based selective harvesting.
        until an optional argument with a UTCdatetime value , which specifies a upper bound for _datestamp-based selective harvesting.
        metadataPrefix a required argument, which specifies that headers should be returned only if the metadata format matching the supplied metadataPrefix is available or, depending on the repository's support for deletions, has been deleted. The metadata formats supported by a repository and for a particular item can be retrieved using the ListMetadataFormats request.
        set an optional argument with a setSpec value , which specifies set criteria for selective harvesting.
        resumptionToken an exclusive argument with a value that is the flow control token returned by a previous ListIdentifiers request that issued an incomplete list.
        Error and Exception Conditions
        badArgument - The request includes illegal arguments or is missing required arguments.
        badResumptionToken - The value of the resumptionToken argument is invalid or expired.
        cannotDisseminateFormat - The value of the metadataPrefix argument is not supported by the repository.
        noRecordsMatch- The combination of the values of the from, until , and set arguments results in an empty list.
        noSetHierarchy - The repository does not support sets.
        */
        listIdentifiersOrRecords(request);
    }

    public void getRecord(OAIRequest request) throws SAXException {
        /*
        Arguments
        identifier a required argument that specifies the unique identifier  of the item in the repository from which the record must be disseminated.
        metadataPrefix a required argument that specifies the metadataPrefix of the format that should be included in the metadata part of the returned record .
        - A record should only be returned if the format specified by the metadataPrefix can be disseminated from the item identified by the value of the identifier argument.
         -The metadata formats supported by a repository and for a particular record can be retrieved using the ListMetadataFormats request.
        Error and Exception Conditions
        badArgument - The request includes illegal arguments or is missing required arguments.
        cannotDisseminateFormat - The value of the metadataPrefix argument is not supported by the item identified by the value of the identifier argument
        idDoesNotExist - The value of the identifier argument is unknown or illegal in this repository.
        */
        String mdPrefix = request.getMetadataPrefix();
        if (!Utilities.checkString(mdPrefix) || !metadataFormats.containsKey(mdPrefix)) return;
        String id = request.getIdentifier();
        Hits hits = executeIdQuery(request);
        if (hits == null || hits.length() == 0) {
            new OAIError(OAIError.ERROR_ID_DOES_NOT_EXIST, "The value of the identifier argument, " + id + ", is unknown or illegal in this repository.").toSAX(this);
            return;//idDoesNotExist
        } else if (hits.length() == 1) {
            try {
                ArrayList availableFormats = verifyMetadataFormatForRecord(hits.doc(0), mdPrefix);
                if (availableFormats != null && !availableFormats.contains(this.metadataFormats.get(mdPrefix)))
                    new OAIError(OAIError.ERROR_CANNOT_DISSEMINATE_FORMAT, "The value of the metadataPrefix argument, " + mdPrefix + ", is not supported by the item identified by the value of the identifier argument, " + id + " .").toSAX(this);
                else {
                    super.startVerbEvent(request);
                    sendRecord(request, hits.doc(0), mdPrefix);
                    super.endVerbEvent(request);
                }
            } catch (IOException e) {
                throw new SAXException(e.getMessage(), e);
            }
        } // FIXME (MP) : On en a deux, on fait quoi ?
    }


    public void listRecords(OAIRequest request) throws SAXException {
        /*
        Summary and Usage Notes
        This verb is used to harvest records from a repository.  Optional arguments permit selective harvesting of records based on set membership and/or _datestamp. Depending on the repository's support for deletions, a returned header may have a status attribute of "deleted" if a record matching the arguments specified in the request has been deleted. No metadata will be present for records with deleted status.

        Arguments
        from an optional argument with a UTCdatetime value , which specifies a lower bound for _datestamp-based selective harvesting.
        until an optional argument with a UTCdatetime value , which specifies a upper bound for _datestamp-based selective harvesting.
        set an optional argument with a setSpec value , which specifies set criteria for selective harvesting.
        resumptionToken an exclusive argument with a value that is the flow control token returned by a previous ListRecords request that issued an incomplete list.
        metadataPrefix a required argument (unless the exclusive argument resumptionToken is used) that specifies the metadataPrefix of the format that should be included in the metadata part of the returned records.   Records should be included only for items from which the metadata format
        matching the metadataPrefix can be disseminated. The metadata formats supported by a repository and for a particular item can be retrieved using the ListMetadataFormats request.
        Error and Exception Conditions
        badArgument - The request includes illegal arguments or is missing required arguments.
        badResumptionToken - The value of the resumptionToken argument is invalid or expired.
        cannotDisseminateFormat - The value of the metadataPrefix argument is not supported by the repository.
        noRecordsMatch - The combination of the values of the from, until, set, and metadataPrefix  arguments results in an empty list.
        noSetHierarchy - The repository does not support sets.
        */
        listIdentifiersOrRecords(request);
    }
    
    /**Protected method which search documents in the LuceneDocumentBase and constructs OAI records.
     * The method to search documents is the same for the verbs ListIdentifiers and ListRecords.
     * 
     * @params request	The request of the harvester.
     * @author Malo Pichot*/
    protected void listIdentifiersOrRecords(OAIRequest request) throws SAXException {
        int verb = request.getVerb();
        String mdPrefix = request.getMetadataPrefix();
        this.resumptionToken = request.getResumptionToken();
        this.newResumptionToken = "";
        int currentBatch = 0;
        int cursor = -1;
        Hits hits = null;
        
        if (Utilities.checkString(resumptionToken)){
            //we have a resumptionToken
            
            if (!_database.entityExists(resumptionToken)){//miss the DBE, we fail
                new OAIError(OAIError.ERROR_BAD_RESUMPTION_TOKEN, "The value of the resumptionToken argument is invalid or expired.").toSAX(this);
                return;
            }
            
            else {
                //we have the databaseEntity, we can go on
                mdPrefix = getResumptionTokenProperty(resumptionToken, OAIRequest.URL_PARAM_NAME_METADATA_PREFIX);
                
            	cursor = Integer.parseInt(getResumptionTokenCursor(resumptionToken));
            	currentBatch = (cursor/this.numRecordsPerResponse)+1;
                
                try {//Get the results in context
                    hits = (Hits) _context.get(/*resultsId*/getResumptionTokenProperty(resumptionToken, "resultsId"));
                } catch (ContextException e) {
                	OAIUtilities.logException(logger, e);
                    new OAIError(OAIError.ERROR_BAD_RESUMPTION_TOKEN, "The value of the resumptionToken argument is invalid or expired.").toSAX(this);
                    resetInternalObjects(resumptionToken, newResumptionToken);
                }
                
            }
            
        } else{//we do not have resumptionToken
            //Check for errors first
            if (!Utilities.checkString(mdPrefix)) {
                new OAIError(OAIError.ERROR_NO_METADATA_FORMATS, "The request does not contain any metadataPrefix").toSAX(this);
                resetInternalObjects(resumptionToken, newResumptionToken);
                return;
            } else if(!metadataFormats.containsKey(mdPrefix)){
                new OAIError(OAIError.ERROR_NO_METADATA_FORMATS, "The metadataPrefix is not available: \""+mdPrefix+"\"").toSAX(this);
                resetInternalObjects(resumptionToken, newResumptionToken);
                return;
            }
            
            try {
                hits = executeQueryForRequestParams(request);//execute the query
            } catch (SDXException e) {
            	OAIUtilities.logException(logger, e);
                resetInternalObjects(resumptionToken, newResumptionToken);
            }
        }

        //Work on results
        if (hits == null || hits.length() == 0) {//noRecordsMatch
            new OAIError(OAIError.ERROR_NO_RECORDS_MATCH, "The combination of the values of the from \""+request.getFrom()+"\", until \""+request.getUntil()+"\", set \""+request.getSetIdentifier()+"\", and metadataPrefix \""+mdPrefix+"\" arguments results in an empty list.").toSAX(this);
            if ( Utilities.checkString(resumptionToken) | Utilities.checkString(newResumptionToken) )//if we were working with resumtionToken, we have to delete info about in database
            	resetInternalObjects(resumptionToken, newResumptionToken);
            return;
        } else{//we have results
            boolean startVerbEventSent = false;
            int resultsLength = hits.length();
            int end = resultsLength - 1;
            int start = currentBatch * this.numRecordsPerResponse;
            int endBatch = start + (this.numRecordsPerResponse - 1);
            for (int i = start; i < resultsLength && i <= endBatch; i++) {
                try {
                    Document hitDoc = hits.doc(i);
                    if (i == endBatch && i != end) {
                        //last document of the current batch, create a new resumptionToken
                        if (newResumptionToken == null || newResumptionToken.equals("")){
                            try {
                                newResumptionToken = createResumptionToken(request);
                            } catch (SDXException e) {
                            	OAIUtilities.logException(logger, e);
                                resetInternalObjects(resumptionToken, newResumptionToken);
                            }
                        }
                        _context.put(newResumptionToken+"_results", hits);//put the results in context ; so, later, we do not need to execute the query again
                    }
                    if (i == end && Utilities.checkString(resumptionToken)) {//reset all things at the last document
                        resetInternalObjects(resumptionToken, newResumptionToken);
                    }
                    ArrayList avaliableFormats = verifyMetadataFormatForRecord(hitDoc, mdPrefix);
                    if (avaliableFormats != null && avaliableFormats.contains(this.metadataFormats.get(mdPrefix))) {
                        if (!startVerbEventSent) {
                            super.startVerbEvent(request);
                            startVerbEventSent = true;
                        }
                        
                        switch(verb){
	                        case OAIRequest.VERB_LIST_IDENTIFIERS:
	                            sendIdentifier(request, hitDoc);
	                        break;
	                        case OAIRequest.VERB_LIST_RECORDS:
	                            sendRecord(request, hitDoc, mdPrefix);
	                        break;
	                        default:
	                        case OAIRequest.VERB_UNKNOWN:
	                            String verbString = request.getVerbString();
	                        	String errorString = "The verb";
		                        if (OAIUtilities.checkString(verbString)) errorString += ", " + verbString + ",";
		                        errorString += " is unknown or a value was not provided.";
		                        new OAIError(OAIError.ERROR_BAD_VERB, errorString).toSAX(this);
	                        break;
                        }
                    }
                } catch (IOException e) {
                	OAIUtilities.logException(logger, e);
                }
            }
            
            if (startVerbEventSent) {
            	//Do we have to send resumption token informations
                if (OAIUtilities.checkString(newResumptionToken)){//we have to send info about a resumptionToken
                    super.sendResumptionToken(newResumptionToken, getResumptionTokenCursor(newResumptionToken), new String(""+resultsLength));
                } else if (OAIUtilities.checkString(resumptionToken)){//harvest is ending, only send info about cursor and completeListSize
                    super.sendResumptionToken(null, new String(""+resultsLength), new String(""+resultsLength));
                }
                super.endVerbEvent(request);
            } else {
                new OAIError(OAIError.ERROR_NO_RECORDS_MATCH, "The combination of the values of the from \""+request.getFrom()+"\", until \""+request.getUntil()+"\", set \""+request.getSetIdentifier()+"\", and metadataPrefix \""+mdPrefix+"\" arguments results in an empty list.").toSAX(this);
                resetInternalObjects(resumptionToken, newResumptionToken);
                return;
            }
        }
    }
    
    //Delete DBE and reset newResumptionToken
    protected void resetInternalObjects (String rt, String newrt){
        String currentRt = rt;
        if (currentRt==null || currentRt.equals(""))
            currentRt = newrt;
        if ( currentRt!=null && !currentRt.equals("")) {//we have to be sure that we have to delete something
	        try {//reset the DBE
	            String resultsId = getResumptionTokenProperty(currentRt, "resultsId");
	            if (resultsId != null) _context.put(resultsId,null);//reset resultsId in context
	            super._database.delete(new DatabaseEntity(currentRt));//delete the DBE
	            super._database.optimize();
	            newrt = null;//no resumptionToken needed
	        } catch (SDXException e) {
	        	OAIUtilities.logException(logger, e);
	        }
        }
    }

    public void listMetadataFormats(OAIRequest request) throws SAXException {
        /*
        Summary and Usage Notes
        This verb is used to retrieve the metadata formats available from a repository.  An optional argument restricts the request to the formats available for a specific item.

        Arguments
        identifier an optional argument that specifies the unique identifier of the item for which available metadata formats are being requested.  If this argument is omitted, then the response includes all metadata formats supported by this repository.  Note that the fact that a metadata format is supported by a repository does not mean that it can be disseminated from all items in the repository.
        Error and Exception Conditions
        badArgument - The request includes illegal arguments or is missing required arguments.
        idDoesNotExist - The value of the identifier argument is unknown or illegal in this repository.
        noMetadataFormats - There are no metadata formats available for the specified item.
        */
        /*output should look like this for each format
        <metadataFormat>
             <metadataPrefix>oai_dc</metadataPrefix>
             <schema>http://www.openarchives.org/OAI/2.0/oai_dc.xsd
               </schema>
             <metadataNamespace>http://www.openarchives.org/OAI/2.0/oai_dc/
               </metadataNamespace>
           </metadataFormat>
          */

        /*NOTE: we don't use the request object here to get an identifier because
        *based upon our architecture we can provide all formats via a pipeline transformation
        * but what if a document doesn't contain any of the mapped fields or a field
        */

        //if we don't have an id we show all mdformates
        if (request != null) {
            String id = request.getIdentifier();
            if (!Utilities.checkString(id)) {
                Enumeration mdformats = this.metadataFormats.elements();
                if (mdformats != null) {
                    super.startVerbEvent(request);
                    while (mdformats.hasMoreElements()) {
                        BasicOAIMetadataFormat mdformat = (BasicOAIMetadataFormat) mdformats.nextElement();
                        if (mdformat != null)
                            mdformat.toSAX(this);
                    }
                    super.endVerbEvent(request);
                }
            } else {
                Hits hits = executeIdQuery(request);
                if (hits == null || hits.length() == 0) {
                    new OAIError(OAIError.ERROR_ID_DOES_NOT_EXIST, "The value of the identifier argument, " + id + ", is unknown or illegal in this repository.").toSAX(this);
                    return;//idDoesNotExist
                } else if (hits.length() == 1) {
                    try {
                        ArrayList availableFormats = verifyMetadataFormatForRecord(hits.doc(0));
                        if (availableFormats == null || availableFormats.size() <= 0)
                            new OAIError(OAIError.ERROR_CANNOT_DISSEMINATE_FORMAT, "There are no metadata formats available for the specified item, " + id + " .").toSAX(this);
                        else {
                            super.startVerbEvent(request);
                            for (int i = 0; i < availableFormats.size(); i++) {
                                BasicOAIMetadataFormat mdFormat = (BasicOAIMetadataFormat) availableFormats.get(i);
                                if (mdFormat != null)
                                    mdFormat.toSAX(this);
                            }
                            super.endVerbEvent(request);
                        }
                    } catch (IOException e) {
                        throw new SAXException(e.getMessage(), e);
                    }
                }
            }

        }


    }

    /**Verifies that a document/record is available in the oai_dc format
     *
     * @param document The document verify
     * @return an ArrayList of the available formats
     */
    protected ArrayList verifyMetadataFormatForRecord(Document document) {
        return verifyMetadataFormatForRecord(document, null);
    }

    /**Verifies that a document/record is available in the specified format
     *
     * @param document The document verify
     * @param mdPrefix The format to verify
     * @return an ArrayList of the available formats
     */
    protected ArrayList verifyMetadataFormatForRecord(Document document, String mdPrefix) {
        ArrayList availableFormats = null;

        if (document != null) {
            Enumeration formatsToCheck = null;
            if (Utilities.checkString(mdPrefix)) {
                BasicOAIMetadataFormat metadataFormat = (BasicOAIMetadataFormat) this.metadataFormats.get(mdPrefix);
                Vector format = new Vector();
                format.add(metadataFormat);
                formatsToCheck = format.elements();
            } else
                formatsToCheck = metadataFormats.elements();


            if (formatsToCheck != null) {
                availableFormats = new ArrayList();
                while (formatsToCheck.hasMoreElements()) {
                    BasicOAIMetadataFormat formatToCheck = (BasicOAIMetadataFormat) formatsToCheck.nextElement();
                    if (formatToCheck != null) {
                        if (Framework.SDXNamespacePrefix.equalsIgnoreCase(formatToCheck.getPrefix()))
                            availableFormats.add(formatToCheck);
                        else {
                            Pipeline pipe = formatToCheck.getPipeline();
                            Parameters mappings = formatToCheck.getMappings();
                            boolean metadataFormatHasMappings = true;
                            if (mappings == null || mappings.getNames() == null && mappings.getNames().length == 0)
                                metadataFormatHasMappings = false;//thie metadataFormat has no mappings so we will send the sdx fields to its pipeline
                            if (pipe != null) {
                                availableFormats.add(formatToCheck);
                            } else if (metadataFormatHasMappings) {
                                String[] sdxFieldNames = mappings.getNames();
                                if (sdxFieldNames != null) {
                                    for (int i = 0; i < sdxFieldNames.length; i++) {
                                        String sdxFieldName = sdxFieldNames[i];
                                        if (Utilities.checkString(sdxFieldName) && !sdxFieldName.equals(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDXDOCID)) {
                                            if (document.getField(sdxFieldName) != null || document.getField(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDX_OAI_DELETED_RECORD) != null) {
                                                availableFormats.add(formatToCheck);
                                            }
                                        }
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }

        return availableFormats;

    }
    
    public String getResumptionToken(){
    	return this.newResumptionToken;
    }

    /**Builds a search locations object
     * using the underlying LuceneIndex
     */
    protected SearchLocations getSearchLocation() {
        SearchLocations sLocs = new SearchLocations();
        sLocs.enableLogging(this.logger);
        sLocs.setUp(this.documentBase); // using new method
        return sLocs;
    }

    /**Builds a basic boolean query
     * adding the global inclusion
     * and exclusion queries.
     */
    
    protected BooleanQuery getBaseQuery() {
        BooleanQuery bq = Utilities.newBooleanQuery();

        if (this.includeQuery != null) {
        	bq.add(this.includeQuery.getLuceneQuery(), BooleanClause.Occur.MUST);
        }
        if (this.excludeQuery != null) {
        	bq.add(this.excludeQuery.getLuceneQuery(), BooleanClause.Occur.MUST_NOT);
        }
        bq.add(
        		new WildcardQuery( new Term(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDX_OAI_DELETED_RECORD, "*") ), 
        		BooleanClause.Occur.SHOULD
		);

        return bq;
    }
    
    protected BooleanQuery getBaseQueryForResults() {
    	BooleanQuery bq = Utilities.newBooleanQuery();;
    	/*request for the oai deleted records*/
    	WildcardQuery oaiDeleted = new WildcardQuery(new Term(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDX_OAI_DELETED_RECORD, "*"));
    	/*request for all docs*/
    	org.apache.lucene.search.Query query = new TermQuery(new Term(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDXALL, "1"));
    	/*if no include and no exclude, we only return deleted*/ 
    	if (this.includeQuery == null && this.excludeQuery == null){
    		bq.add (query, BooleanClause.Occur.SHOULD);
    		bq.add (oaiDeleted, BooleanClause.Occur.SHOULD);
    	 }
    	else {
    		if (this.includeQuery != null && this.excludeQuery == null){
    			// query between include and exclude but here no exclude
    			BooleanQuery bq2 = Utilities.newBooleanQuery();
    			
    			bq2.add(this.includeQuery.getLuceneQuery(), BooleanClause.Occur.MUST);
    			bq.add(bq2,BooleanClause.Occur.SHOULD);			// query with include and exclude
    			bq.add(oaiDeleted, BooleanClause.Occur.SHOULD);	// query with deleted
    		}
    		else if (this.excludeQuery != null && this.includeQuery == null){
    			// query between include and exclude but here no include
    			BooleanQuery bq2 = Utilities.newBooleanQuery();
    			bq2.add(query, BooleanClause.Occur.MUST);
    			bq2.add(this.excludeQuery.getLuceneQuery(), BooleanClause.Occur.MUST_NOT);
    			bq.add(bq2,BooleanClause.Occur.SHOULD);
    			bq.add(oaiDeleted, BooleanClause.Occur.SHOULD);	// query with deleted
    		}
    		else if (this.excludeQuery != null && this.includeQuery != null){
    			// We want only include and except exclude
    			BooleanQuery bq2 = Utilities.newBooleanQuery();
    			bq2.add(this.includeQuery.getLuceneQuery(), BooleanClause.Occur.MUST);
    			bq2.add(this.excludeQuery.getLuceneQuery(), BooleanClause.Occur.MUST_NOT);
    			bq.add(bq2,BooleanClause.Occur.SHOULD);
    			bq.add(oaiDeleted, BooleanClause.Occur.SHOULD);	// query with deleted
    		}
    	}
    	
    	return bq;
    }


    protected TermQuery getIdQuery(String id) {
        TermQuery tq = null;
        if (Utilities.checkString(id)) {
            tq = new TermQuery(new Term(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDXDOCID, id));
        }
        return tq;
    }

    /**Executes a query from an OAI request for the identifier param*/
    protected Hits executeIdQuery(OAIRequest request) throws SAXException {
        String id = request.getIdentifier();
        String internalId = deriveInternalSdxId(id);
        Hits hits = null;
        if (Utilities.checkString(internalId)) {
            try {
                BooleanQuery bq = getBaseQueryForResults();
                TermQuery idQuery = getIdQuery(internalId);
                bq.add(idQuery, BooleanClause.Occur.MUST);

                if (this.luceneSearchIndex != null)
                    hits = luceneSearchIndex.search(bq);
                OAIUtilities.logDebug(logger, "Executing an ID query: "+idQuery.toString()+" ; OAI reposiotiry: "+this.getId()+" ; document base: "+this.documentBaseId+" ; results: "+hits.length());
            } catch (SDXException e) {
                throw new SAXException(e.getMessage(), e);
            }

        }
        return hits;
    }


    //TODO: use DocumentBaseOAIRecord
    protected void sendRecord(OAIRequest request, org.apache.lucene.document.Document doc, String mdPrefix) throws SAXException {
        /*
     <record>
        <header>
          <identifier>oai:arXiv.org:cs/0112017</identifier>
          <datestamp>2001-12-14</datestamp>
          <setSpec>cs</setSpec>
          <setSpec>math</setSpec>
        </header>
        <metadata>
          <oai_dc:dc
             xmlns:oai_dc="http://www.openarchives.org/OAI/2.0/oai_dc/"
             xmlns:dc="http://purl.org/dc/elements/1.1/"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://www.openarchives.org/OAI/2.0/oai_dc/
             http://www.openarchives.org/OAI/2.0/oai_dc.xsd">
            <dc:title>Using Structural Metadata to Localize Experience of
                      Digital Content</dc:title>
            <dc:creator>Dushay, Naomi</dc:creator>
            <dc:subject>Digital Libraries</dc:subject>
            <dc:description>With the increasing technical sophistication of
                both information consumers and providers, there is
                increasing demand for more meaningful experiences of digital
                information. We present a framework that separates digital
                object experience, or rendering, from digital object storage
                and manipulation, so the rendering can be tailored to
                particular communities of users.
            </dc:description>
            <dc:description>Comment: 23 pages including 2 appendices,
                8 figures</dc:description>
            <dc:date>2001-12-14</dc:date>
          </oai_dc:dc>
        </metadata>
      </record>
     */

        BasicOAIMetadataFormat metadataFormat = (BasicOAIMetadataFormat) this.metadataFormats.get(mdPrefix);
        String formatPrefix = metadataFormat.getPrefix();
        boolean isSDXFormat = formatPrefix.equalsIgnoreCase(Framework.SDXNamespacePrefix);
        if (doc != null) {
            if (doc.getField(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDXDOCID) == null || doc.getField(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDX_OAI_DELETED_RECORD) != null){
            	super.startElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.RECORD, OAIObject.Node.Name.RECORD, null);
            	sendHeaderEvents(request, doc);
            	super.endElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.RECORD, OAIObject.Node.Name.RECORD);
            }
            else if (metadataFormat != null) {
                //start element 'record'
                super.startElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.RECORD, OAIObject.Node.Name.RECORD, null);

                sendHeaderEvents(request, doc);

                // TODO (MP): utiliser doc.getFields() qui renvoie une List.
                Enumeration fields = doc.fields();
                XMLPipe pipe = null;
                boolean metadataFormatHasMappings = true;
                if (fields != null) {
                    boolean startEventsSent = false;
                    while (fields.hasMoreElements()) {
                        if (metadataFormat.getMappings() == null || metadataFormat.getMappings().getNames() == null
                                && metadataFormat.getMappings().getNames().length == 0)
                            metadataFormatHasMappings = false;//thie metadataFormat has no mappings so we will send the sdx fields to its pipeline
                        pipe = metadataFormat.getPipeline();
                        if (pipe != null) {
                            Pipeline addRootAttPipe = metadataFormat.getAddRootAttributePipe();
                            if (addRootAttPipe != null)
                                addRootAttPipe.setConsumer(new IncludeXMLConsumer(this));
                            pipe.setConsumer(addRootAttPipe);
                        }
                        if (!startEventsSent) {
                            super.startElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.METADATA, OAIObject.Node.Name.METADATA, null);
                            metadataFormat.setConsumer(this);
                            if (metadataFormatHasMappings || isSDXFormat)//we don't have a pipe so each mapped element needs to be contained within a rootElement
                                metadataFormat.startMetadataFormatRootElement();
                            startEventsSent = true;
                        }
                        Field field = (Field) fields.nextElement();
                        if (field != null) {
                            String origName = field.name();
                            String val = field.stringValue();
                            // Transform date if a date field...
                            if (this.luceneSearchIndex.getFieldType(origName) == fr.gouv.culture.sdx.search.lucene.Field.DATE)
                                val = fr.gouv.culture.sdx.utils.Date.formatDate(fr.gouv.culture.sdx.search.lucene.DateField.stringToDate(val));
                            if (Utilities.checkString(origName)) {
                                try {
                                    String newVal = val;
                                    //if there is a pipeline and no field mappings we get the document and send it to the pipe
                                    if (origName.equals(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDXDOCID)) {
                                        if (pipe != null && !metadataFormatHasMappings) {
                                            XMLDocument xdoc = new XMLDocument();
                                            xdoc.enableLogging(this.logger);
                                            xdoc.setId(val);
                                            this.documentBase.getDocument(xdoc, pipe, false);
                                        } else if (isSDXFormat)
                                            metadataFormat.sendElement(origName, newVal);
                                        else {//we have mappings and so we build and external representation TODO: is this a good idea in all cases?
                                            newVal = buildUrlLocator(request, val);
                                            metadataFormat.sendElement(origName, newVal);
                                        }
                                    } else {
                                        //if anything was mapped we send the elements from the mappings
                                        metadataFormat.sendElement(origName, newVal);
                                    }
                                } catch (SDXException e) {
                                	OAIUtilities.logException(logger, e);
                                }

                            }
                        }
                    }

                    if (startEventsSent) {
                        if (metadataFormatHasMappings || isSDXFormat)//we don't have a pipe so each mapped element needs to be contained within a rootElement
                            metadataFormat.endMetadataFormatRootElement();
                        super.endElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.METADATA, OAIObject.Node.Name.METADATA);
                    }

                }

                super.endElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.RECORD, OAIObject.Node.Name.RECORD);
            }
        }


    }


    protected void sendIdentifier(OAIRequest request, org.apache.lucene.document.Document doc) throws SAXException {
        /*
     <record>
        <header>
          <identifier>oai:arXiv.org:cs/0112017</identifier>
          <datestamp>2001-12-14</datestamp>
          <setSpec>cs</setSpec>
          <setSpec>math</setSpec>
        </header>
        <metadata>
          <oai_dc:dc
             xmlns:oai_dc="http://www.openarchives.org/OAI/2.0/oai_dc/"
             xmlns:dc="http://purl.org/dc/elements/1.1/"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://www.openarchives.org/OAI/2.0/oai_dc/
             http://www.openarchives.org/OAI/2.0/oai_dc.xsd">
            <dc:title>Using Structural Metadata to Localize Experience of
                      Digital Content</dc:title>
            <dc:creator>Dushay, Naomi</dc:creator>
            <dc:subject>Digital Libraries</dc:subject>
            <dc:description>With the increasing technical sophistication of
                both information consumers and providers, there is
                increasing demand for more meaningful experiences of digital
                information. We present a framework that separates digital
                object experience, or rendering, from digital object storage
                and manipulation, so the rendering can be tailored to
                particular communities of users.
            </dc:description>
            <dc:description>Comment: 23 pages including 2 appendices,
                8 figures</dc:description>
            <dc:date>2001-12-14</dc:date>
          </oai_dc:dc>
        </metadata>
      </record>
     */



        if (doc != null)
            sendHeaderEvents(request, doc);

    }


    /**Sends the OAI-PMH header events for the provided lucene document
     *
     * @param request The request
     * @param doc   The lucene document
     * @throws SAXException
     */
    //TODO: use DocumentBaseOAIRecord
    protected void sendHeaderEvents(OAIRequest request, Document doc) throws SAXException {

        String id = doc.get(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDXDOCID);
        String externalId = "";
        String status = null;
        if (!Utilities.checkString(id)) {
            String deletedId = doc.get(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDX_OAI_DELETED_RECORD);
            if (Utilities.checkString(deletedId)) {
                status = OAIObject.Node.Value.DELETED;
                id = deletedId;
            }
        }
        //externalId = buildExternalOaiId(request, id);
        externalId = buildExternalOaiId(id);
        String datestamp = doc.get(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDXMODDATE);


        AttributesImpl atts = new AttributesImpl();
        if (Utilities.checkString(status))
            atts.addAttribute("", OAIObject.Node.Name.STATUS, OAIObject.Node.Name.STATUS, OAIObject.Node.Type.CDATA, status);

        super.startElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.HEADER, OAIObject.Node.Name.HEADER, atts);

        datestamp = fr.gouv.culture.sdx.utils.Date.formatUtcISO8601Date(DateField.stringToDate(datestamp));

        super.sendElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.IDENTIFIER, OAIObject.Node.Name.IDENTIFIER, null, externalId);
        super.sendElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.DATESTAMP, OAIObject.Node.Name.DATESTAMP, null, datestamp);
        if (this.setMappings != null) {
            Enumeration sets = this.setMappings.keys();
            if (sets != null) {
                while (sets.hasMoreElements()) {
                    Parameters setParams = getDocumentSetParameters(doc);
                    if (setParams != null) {
                        String setSpec = setParams.getParameter(PARAMETER_NAME_SET_SPEC, "");
                        if (Utilities.checkString(setSpec)) {
                            String encoding = Utilities.getStringFromContext(ContextKeys.SDX.ENCODING, getContext());
                            setSpec = Utilities.encodeURL(setSpec, encoding);
                            super.sendElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.SET_SPEC, OAIObject.Node.Name.SET_SPEC, null, setSpec);
                        }
                    }
                }
            }
        }

        super.endElement(OAIObject.Node.Xmlns.OAI_2_0, OAIObject.Node.Name.HEADER, OAIObject.Node.Name.HEADER);

    }


    /**Adds a deleted record lucene document
     * to the search index for the specific id
     *
     * @param id    The id of the deleted record
     * @throws SDXException
     */
    public void addDeletedRecord(String id) throws SDXException {
        if (!Utilities.checkString(id)) return;
        OAIUtilities.logDebug(this.logger, "Adding the doc "+id+" from the list of deleted docs of the OAI repository "+getId());
        String docId = id;
        org.apache.lucene.document.Document deletedRecord = new org.apache.lucene.document.Document();
        Hits hits = null;
        BooleanQuery bq = getBaseQuery();
        bq.add(getIdQuery(docId), BooleanClause.Occur.MUST);
        hits = this.luceneSearchIndex.search(bq);

        try {
            if (hits != null && hits.length() == 1 && docId.equals(hits.doc(0).get(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDXDOCID))) {
            	deletedRecord.add(
            			new Field(	LuceneDocumentBase.INTERNAL_FIELD_NAME_SDX_OAI_DELETED_RECORD, id, 
            						Field.Store.YES, Field.Index.UN_TOKENIZED, Field.TermVector.NO
        						)
            	);
                Date date = fr.gouv.culture.sdx.utils.Date.getUtcIso8601Date();
                deletedRecord.add(
                		new Field(	LuceneDocumentBase.INTERNAL_FIELD_NAME_SDXMODDATE, DateField.dateToString(date),
                					Field.Store.YES, Field.Index.UN_TOKENIZED, Field.TermVector.NO
                				)
                );
                luceneSearchIndex.writeDocument(deletedRecord, false);//passing false here as a calling method should be responsible for any index optimization.
            }
        } catch (IOException e) {
        	OAIUtilities.logException(logger, e);//TODO: Exception, can we do anything better here
        }

    }

    /**Removes a deleted record entry from lucene document
     * from the search index for the specific id
     *
     * @param id    The id of the deleted record
     * @throws SDXException
     */
    public void removeDeletedRecord(String id) throws SDXException {
        if (!Utilities.checkString(id)) return;
        OAIUtilities.logDebug(this.logger, "Removing the doc "+id+" from the list of deleted docs of the OAI repository "+getId());
        Term term = new Term(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDX_OAI_DELETED_RECORD, id);
        luceneSearchIndex.delete(term);
    }

    /**Returns  a boolean query for a specific set for a specific set
     *
     * @param setSpec   The setSpec to use
     */
    protected BooleanQuery addSetQuery(String setSpec) {
        BooleanQuery bq = Utilities.newBooleanQuery();
        if (this.setMappings != null && Utilities.checkString(setSpec)) {
            Parameters setParams = (Parameters) this.setMappings.get(setSpec);
            if (setParams != null) {
                String sdxFieldName = setParams.getParameter(PARAMETER_NAME_SDX_FIELD, "");
                String includeQuery = setParams.getParameter(PARAMETER_NAME_INCLUDE_QUERY, "");
                String excludeQuery = setParams.getParameter(PARAMETER_NAME_EXCLUDE_QUERY, "");
                if (Utilities.checkString(includeQuery) || Utilities.checkString(excludeQuery)) {
                    try {
                        org.apache.lucene.search.Query inQuery = getLuceneSimpleQuery(includeQuery);
                        if (inQuery != null)
                        	bq.add(inQuery, BooleanClause.Occur.MUST);
                    } catch (SDXException e) {
                        //this exception should never be thrown, as configuration will fail before if the query is unparsable
                    	OAIUtilities.logException(logger, e);
                    }

                    try {
                        org.apache.lucene.search.Query exQuery = getLuceneSimpleQuery(excludeQuery);
                        if (exQuery != null)
                        	bq.add(exQuery, BooleanClause.Occur.MUST_NOT);
                    } catch (SDXException e) {
                        //this exception should never be thrown, as configuration will fail before if the query is unparsable
                    	OAIUtilities.logException(logger, e);
                    }

                } else if (Utilities.checkString(sdxFieldName))
                	bq.add(	new WildcardQuery( new Term(sdxFieldName, setSpec) ), BooleanClause.Occur.MUST );
            }
        }
        return bq;
    }

    /**Builds an sdx simple query for the provided query string
     * and returns the underlying lucene query object
     *
     * @param query The query string
     * @throws SDXException
     */
    protected org.apache.lucene.search.Query getLuceneSimpleQuery(String query) throws SDXException {
        org.apache.lucene.search.Query ret = null;
        if (Utilities.checkString(query)) {
            SearchLocations sLocs = this.getSearchLocation();
            SimpleQuery sQuery = new SimpleQuery();
            sQuery.enableLogging(this.logger);
            sQuery.setUp(sLocs, null, query);
            ret = sQuery.getLuceneQuery();
        }
        return ret;
    }


    /**Returns a parameters object containing
     * set information for a lucene document
     *
     * @param doc   The lucene document
     */
    protected Parameters getDocumentSetParameters(Document doc) {
        Parameters ret = null;
        String docId = doc.get(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDXDOCID);
        Enumeration setSpecs = this.setMappings.keys();
        if (setSpecs != null) {
            while (setSpecs.hasMoreElements()) {
                String setSpec = (String) setSpecs.nextElement();
                Parameters setParams = (Parameters) this.setMappings.get(setSpec);
                String sdxFieldName = setParams.getParameter(PARAMETER_NAME_SDX_FIELD, "");
                if (Utilities.checkString(sdxFieldName)) {
                    Field field = doc.getField(sdxFieldName);
                    if (field != null) {
                        String fieldValue = field.stringValue();
                        //checking if the setSpec indicates a field name or its value
                        if (Utilities.checkString(setSpec) && setSpec.equals(sdxFieldName) || setSpec.equals(fieldValue)) {
                            ret = setParams;
                            break;
                        }
                    }
                }
                // if the above doesn't work then we will evaluate the inlcude/exclude queries
                BooleanQuery bq = getBaseQueryForResults();
                String includeQuery = setParams.getParameter(PARAMETER_NAME_INCLUDE_QUERY, "");
                String excludeQuery = setParams.getParameter(PARAMETER_NAME_EXCLUDE_QUERY, "");
                if (Utilities.checkString(includeQuery) || Utilities.checkString(excludeQuery)) {
                    try {
                        org.apache.lucene.search.Query inQuery = getLuceneSimpleQuery(includeQuery);
                        if (inQuery != null)
                        	bq.add(inQuery, BooleanClause.Occur.MUST);
                    } catch (SDXException e) {
                        //this exception should never be thrown, as configuration will fail before if the query is unparsable
                    	OAIUtilities.logException(logger, e);
                    }
                    try {
                        org.apache.lucene.search.Query exQuery = getLuceneSimpleQuery(excludeQuery);
                        if (exQuery != null)
                        	bq.add(exQuery, BooleanClause.Occur.MUST_NOT);
                    } catch (SDXException e) {
                        //this exception should never be thrown, as configuration will fail before if the query is unparsable
                    	OAIUtilities.logException(logger, e);
                    }
                    //now execute the query and then compare the docids
                    if (this.luceneSearchIndex != null) {
                        try {
                            Hits hits = luceneSearchIndex.search(bq);
                            if (hits != null && hits.length() > 0) {
                                for (int i = 0; i < hits.length(); i++) {
                                    Document hitDoc = hits.doc(i);
                                    String id = hitDoc.get(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDXDOCID);
                                    if (Utilities.checkString(id) && Utilities.checkString(docId) && id.equals(docId)) {
                                        ret = setParams;
                                        break;
                                    }
                                }
                            }
                        } catch (SDXException e) {
                        	OAIUtilities.logException(logger, e);
                        } catch (IOException e) {
                        	OAIUtilities.logException(logger, e);
                        }
                    }


                }


            }


        }


        return ret;
    }


    /** Executes a query for a request and sorts them in
     * ascending date order
     *
     * @param request The request from which to build the query to execute
     * @return Hits		
     * @throws SDXException
     * @throws SAXException
     */
    protected Hits executeQueryForRequestParams(OAIRequest request) throws SDXException, SAXException {
        String set = "";
        String from = "";
        String until = "";
        if (request != null) {
            set = request.getSetIdentifier();
            resumptionToken = request.getResumptionToken();
            from = request.getFrom();
            until = request.getUntil();
        }

        BooleanQuery bq = getBaseQueryForResults();
        
        org.apache.lucene.search.Filter rangeFilter = null;
        java.util.Date fromDate = null;
        java.util.Date untilDate = null;
        
        //Validation of set hierarchy
        if (Utilities.checkString(set) && (this.setMappings == null || this.setMappings.size() == 0)) {
            super.sendNoSetHierarchyError();
            return null;
        }

        /*Start Date validation*/
        // les dates ont deja ete validees au moment de la reception de la requete ; inutile de le faire a nouveau ici ! [MP]
        //from param
        if (fromDate == null){//we check for null, as we may be using a date from the last record in a resumptionToken set
            fromDate = fr.gouv.culture.sdx.utils.Date.parseUtcISO8601DateDayOrSecond(from);
        }
        untilDate = fr.gouv.culture.sdx.utils.Date.parseUtcISO8601DateDayOrSecond(until);
        /*End Date validation*/
        if (fromDate != null || untilDate != null) {
            
            if (fromDate != null && untilDate != null) {
                /*see OAI 2.0 spec 2.7.1 Selective Harvesting and Datestamps :
                * Therefore, the from arugment must be less than or equal to the until argument.
                *Otherwise, a repository must issue a badArgument error .
                */
                if (fromDate.getTime() > untilDate.getTime()) {
                    new OAIError(OAIError.ERROR_BAD_ARGUMENT, "The value of the \"from\" argument, " + from + ", must be less than or equal to the value of the \"until\" argument," + until + " .").toSAX(this);
                    return null;//Bad dates
                }
                else if ( until.equals(from) && getGranularity(until)==OAIObject.Node.Value.STRING_GRANULARITY_DAY
            		// 	from and until are equals for a day granularity
        		) untilDate = fr.gouv.culture.sdx.utils.Date.parseUtcISO8601Date( until + "T23:59:59" );
                	
                // adds 999 milliseconds to the until param to include it in the range
                untilDate.setTime( untilDate.getTime() + 999 );
            }
            
            OAIUtilities.logDebug(
            		logger, 
            		"from="+fr.gouv.culture.sdx.utils.Date.formatDate( fromDate, "yyyy-MM-dd'T'HH:mm:ss:SZ") +
            		"\tuntil="+fr.gouv.culture.sdx.utils.Date.formatDate( untilDate, "yyyy-MM-dd'T'HH:mm:ss:SZ")
            		, null);
            
            DateIntervalQuery dateQuery = new DateIntervalQuery();
            dateQuery.enableLogging(this.logger);
            dateQuery.setUpOai(getSearchLocation(), LuceneDocumentBase.INTERNAL_FIELD_NAME_SDXMODDATE, fromDate, untilDate);
            rangeFilter = dateQuery.prepare();
        }
            

        bq.add(addSetQuery(set), BooleanClause.Occur.SHOULD);
        
        Hits hits = null;
        if (this.luceneSearchIndex != null)
            hits = luceneSearchIndex.search(bq, rangeFilter);

		return hits;

    }


    /**Returns the earliest _datestamp for this repository
     *
     */
    public String getEarliestDatestamp() {
    	if (!Utilities.checkString(super.earliestDatestamp)) {
    		Date utcDate = fr.gouv.culture.sdx.utils.Date.getUtcIso8601Date(this.documentBase.creationDate());
    		super.earliestDatestamp = fr.gouv.culture.sdx.utils.Date.formatUtcISO8601Date(utcDate);
    	}
    	return super.earliestDatestamp;
    }

    /**Destroys all deleted record entries
     *
     */
    public void purgeDeletedRecords() {
        try {
            LuceneIndex luceneSearchIndex = this.luceneSearchIndex;
            luceneSearchIndex.delete(new Term(LuceneDocumentBase.INTERNAL_FIELD_NAME_SDX_OAI_DELETED_RECORD, "*"));
            luceneSearchIndex.optimize();
        } catch (SDXException e) {
        	OAIUtilities.logException(logger, e);
        }

    }


}


