DocumentServiceImpl

package org.kuali.ole.docstore.service;

import org.apache.commons.lang.time.StopWatch;
import org.kuali.ole.RepositoryManager;
import org.kuali.ole.docstore.OleDocStoreException;
import org.kuali.ole.docstore.model.xmlpojo.ingest.Request;
import org.kuali.ole.docstore.model.xmlpojo.ingest.RequestDocument;
import org.kuali.ole.docstore.model.xmlpojo.ingest.Response;
import org.kuali.ole.docstore.model.xmlpojo.ingest.ResponseDocument;
import org.kuali.ole.docstore.process.BulkIngestNIndexProcessor;
import org.kuali.ole.docstore.process.BulkLoadHandler;
import org.kuali.ole.docstore.process.DocStoreCamelContext;
import org.kuali.ole.docstore.process.ProcessParameters;
import org.kuali.ole.docstore.process.batch.BulkProcessRequest;
import org.kuali.ole.docstore.transaction.TransactionManager;
import org.kuali.ole.docstore.utility.BatchIngestStatistics;
import org.kuali.ole.docstore.utility.BulkIngestStatistics;
import org.kuali.ole.pojo.OleException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

/**
 * Implements DocumentService interface.
 * User: tirumalesh.b
 * Date: 28/8/12 Time: 12:09 PM
 */
public class DocumentServiceImpl
        implements DocumentService {
    private static Logger              logger            = LoggerFactory.getLogger(BulkIngestNIndexProcessor.class);
    private static DocumentServiceImpl documentService   = new DocumentServiceImpl();
    private        BulkProcessRequest  bulkIngestRequest = null;
    //    private        BulkLoadHandler              bulkLoadHandler              = null;
    //    private        BulkIngestNIndexRouteBuilder bulkIngestNIndexRouteBuilder = null;
    //    private        BulkIngestNIndexProcessor    bulkIngestNIndexProcessor    = null;
    //    private        BulkIngestStatistics         bulkLoadStatistics           = null;
    //    private BulkIngestProcessHandlerService bIService        = (BulkIngestProcessHandlerService) BeanLocator
    //                                                                     .getBean("bulkIngestProcessHandlerService");
    protected RepositoryManager repositoryManager;

    public static DocumentServiceImpl getInstance() {
        return documentService;
    }

    private DocumentServiceImpl() {
        try {
            this.repositoryManager = RepositoryManager.getRepositoryManager();
        }
        catch (OleException oe) {
            //throw new OleDocStoreException(oe);
            // TODO: log the exception
        }
    }

    /**
     * @inheritDoc
     */
    @Override
    public Response process(Request request) throws OleDocStoreException {
        authenticateAndAuthorizeUser(request);
        validateInput(request);

        TransactionManager transactionManager = new TransactionManager();
        transactionManager.startTransaction(request.getUser(), request.getOperation());
        List<ResponseDocument> responseDocuments = new ArrayList<ResponseDocument>();
        Response response = new Response();
        response.setOperation(request.getOperation());
        response.setUser(request.getUser());
        try {
            if (request.getOperation().equals(Request.Operation.ingest.toString())) {
                responseDocuments = transactionManager.ingest(request.getRequestDocuments());
                transactionManager.commit();

                response.setStatus("Success");
                response.setMessage("Documents ingested.");
                response.setStatusMessage("Documents ingested successfully.");
                response.setDocuments(responseDocuments);
            }

            else if (request.getOperation().equals(Request.Operation.checkIn.toString())) {
                responseDocuments = transactionManager.checkIn(request.getRequestDocuments(), request.getOperation());
                transactionManager.commit();

                response.setStatus("Success");
                response.setMessage("Documents Checked In .");
                response.setStatusMessage("Documents Checked In successfully.");
                response.setDocuments(responseDocuments);
            }
            else if (request.getOperation().equals(Request.Operation.checkOut.toString())) {
                responseDocuments = transactionManager.checkOut(request.getRequestDocuments(), request.getUser());

                response.setStatus("Success");
                response.setMessage("Documents Checked Out.");
                response.setStatusMessage("Documents Checked Out successfully.");
                response.setDocuments(responseDocuments);
            }
            else if (request.getOperation().contains(Request.Operation.delete.toString())) {
                responseDocuments = transactionManager.delete(request.getRequestDocuments());
                transactionManager.commit();
                response.setStatus("Success");
                response.setMessage("Documents deleted");
                response.setStatusMessage("Documents deleted successfully");
                response.setDocuments(responseDocuments);
            }
        }
        catch (OleDocStoreException ode) {
            logger.error("", ode);
            transactionManager.abort();
            response.setStatus("Failure");
            response.setMessage("Operation failed.");
            response.setStatusMessage("Operation failed.");
            response.setDocuments(responseDocuments);
        }

        return response;
    }

    /**
     * @inheritDoc
     */
    @Override
    public void bulkProcess(BulkProcessRequest request) throws Exception {
        authenticateAndAuthorizeUser(request);
        validateBulkProcessInput(request);

        Response response = null;
        if (request.getOperation().equals(BulkProcessRequest.BulkProcessOperation.INGEST)) {
            response = bulkIngestManage(request);
        }
    }

    /**
     * Manages the bulk ingest process.
     */
    private Response bulkIngestManage(BulkProcessRequest request) throws Exception {
        if ((request.getAction()).equals(BulkProcessRequest.BulkProcessAction.START)) {
            if (request.getDataFormat().equals(BulkProcessRequest.BulkIngestDataFormat.DOCSTORE)) {
                if (bulkIngestRequest != null) {
                    //DocStoreCamelContext.getInstance().resume();
                    throw new OleDocStoreException("Bulk ingest already running!");
                }
                else {
                    BeanLocator.getBulkIngestProcessHandlerService().startBulkIngestForDocStoreRequestFormat(null);
                    BulkLoadHandler bulkLoadHandler = BeanLocator.getBulkIngestProcessHandlerService().getLoadHandler();
                    BulkIngestNIndexProcessor bulkIngestNIndexProcessor = bulkLoadHandler.getBulkRoute()
                                                                                         .getBulkIngestNIndexProcessor();
                    request.setBulkIngestStatistics(BulkIngestStatistics.getInstance());
                    bulkIngestNIndexProcessor.setBulkProcessRequest(request);
                    //bulkLoadStatistics = bulkIngestNIndexProcessor.getBulkLoadStatistics();
                    bulkIngestRequest = request;
                    DocStoreCamelContext.getInstance().resume();
                }
            }
            else if (request.getDataFormat().equals(BulkProcessRequest.BulkIngestDataFormat.STANDARD)) {
                String folder = request.getDataFolder();
                if (folder != null && folder.trim().length() != 0) {
                    BeanLocator.getBulkIngestProcessHandlerService()
                               .startBulkIngestForStandardXMLFormat(request.getDataFolder(), request.getDocCategory(),
                                                                    request.getDocType(), request.getDocFormat(),null);
                }
            }
        }
        else if ((request.getAction()).equals(BulkProcessRequest.BulkProcessAction.STATUS)) {
            logger.info(bulkIngestRequest.getBulkIngestStatistics().getJsonString());
        }
        else if ((request.getAction()).equals(BulkProcessRequest.BulkProcessAction.STOP)) {
            DocStoreCamelContext.getInstance().suspend();
        }
        else if ((request.getAction()).equals(BulkProcessRequest.BulkProcessAction.CLEAR)) {
            bulkIngestRequest.getBulkIngestStatistics().clearBulkIngestStatistics();
        }
        return null;
    }

    public void bulkIngest(BulkProcessRequest bulkProcessRequest, List<RequestDocument> requestDocuments)
            throws OleDocStoreException {
        TransactionManager transactionManager = bulkProcessRequest.getTransactionManager();
        if (null == transactionManager) {
            transactionManager = new TransactionManager();
            transactionManager.startTransaction(bulkProcessRequest.getUser(),
                                                "bulk" + bulkProcessRequest.getOperation().toString());
            bulkProcessRequest.setTransactionManager(transactionManager);
        }
        //        if (bulkProcessRequest.getPreviousBatchDocuments() == null) {
        //            List<RequestDocument> newList = new ArrayList<RequestDocument>();
        //            bulkProcessRequest.setPreviousBatchDocuments(newList);
        //        }
        //        bulkProcessRequest.getPreviousBatchDocuments().addAll(requestDocuments);

        batchIngest(bulkProcessRequest, requestDocuments);

    }

    public void batchIngest(BulkProcessRequest bulkProcessRequest, List<RequestDocument> requestDocuments) {
        BulkIngestStatistics bulkIngestStatistics = bulkProcessRequest.getBulkIngestStatistics();
        BatchIngestStatistics batchStatistics = bulkIngestStatistics.getCurrentBatch();
        long commitSize = ProcessParameters.BULK_INGEST_COMMIT_SIZE;
        logger.debug("commitSize = " + commitSize);
        logger.debug("bulkIngestNIndex(" + requestDocuments.size() + ") START");
        logger.debug("BULK_INGEST_IS_LINKING_ENABLED=" + ProcessParameters.BULK_INGEST_IS_LINKING_ENABLED);
        StopWatch totalTimer = new StopWatch();
        long recCount = requestDocuments.size();
        boolean isCommit = false;
        totalTimer.start();
        TransactionManager transactionManager = bulkProcessRequest.getTransactionManager();
        try {
            transactionManager.batchIngest(bulkProcessRequest, requestDocuments);

            bulkIngestStatistics.setCommitRecCount(bulkIngestStatistics.getCommitRecCount() + recCount);
            if (bulkIngestStatistics.getCommitRecCount() == commitSize || bulkIngestStatistics.isLastBatch()) {
                isCommit = true;
            }
            if (isCommit) {
                logger.info("Bulk ingest: Commit started. Number of records being committed : " + bulkIngestStatistics
                        .getCommitRecCount());
                transactionManager.commit();
                bulkIngestStatistics.setCommitRecCount(0);
            }

            // Documents processed can be different from records processed as in the case of Instance data.
            logger.debug("Documents processed: " + recCount);
            bulkIngestStatistics.setFileRecCount(bulkIngestStatistics.getFileRecCount() + recCount);
            logger.info(
                    "Bulk ingest: Records processed in the current file :" + bulkIngestStatistics.getFileRecCount());
        }
        catch (Exception e) {
            transactionManager.abort();
            bulkIngestStatistics.setCommitRecCount(0);
            logger.error("Document Ingest & Index Failed, Cause: " + e.getMessage(), e);
        }
        totalTimer.stop();
        batchStatistics.setIngestingTime(
                batchStatistics.getTimeToCreateNodesInJcr() + batchStatistics.getTimeToSaveJcrSession());
        batchStatistics
                .setIndexingTime(batchStatistics.getTimeToIndexSolrInputDocs() + batchStatistics.getTimeToSolrCommit());
        batchStatistics.setIngestNIndexTotalTime(totalTimer.getTime());
    }


    public void setRepositoryManager(RepositoryManager repositoryManager) {
        this.repositoryManager = repositoryManager;
    }

    /*
     *
     *
     * @param request
     */

    /**
     * Validates the given request for normal processing.
     *
     * @param request
     * @throws OleDocStoreException - if the operation is invalid
     *                              - if no documents are specified
     */
    private void validateInput(Request request) throws OleDocStoreException {
        // TODO: Implement now.
    }

    /**
     * Verifies whether the user is authenticated and authorized to execute the request.
     *
     * @param request
     * @throws OleDocStoreException
     */
    private void authenticateAndAuthorizeUser(Request request) throws OleDocStoreException {
        // TODO: Implement later.
    }

    private void authenticateAndAuthorizeUser(BulkProcessRequest request) throws OleDocStoreException {
        // TODO: Implement later.
    }

    /**
     * Validates the given request for bulk processing.
     *
     * @param request
     * @throws OleDocStoreException - if the operation is invalid
     *                              - if no documents are specified
     *                              - if all documents are not of same [cat-type-format]
     */
    private void validateBulkProcessInput(BulkProcessRequest request) throws OleDocStoreException {
        // TODO: Implement now.
    }

}