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

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.synapse.Mediator;
import org.apache.synapse.MessageContext;
import org.apache.synapse.SynapseException;
import org.apache.synapse.message.processors.MessageProcessorConstants;
import org.apache.synapse.message.processors.ScheduledMessageProcessor;
import org.apache.synapse.message.store.MessageStore;
import org.apache.synapse.util.xpath.SynapseXPath;
import org.quartz.Job;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;

import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * All necessary logic for Resequencing is implemented with in this class.
 * This class extends from Job class which comes from Quartz
 */
public class ResequencingJob implements Job {

    /**
     * Log is set to the current class
     */
    private static final Log log = LogFactory.getLog(ResequencingJob.class);

    /**
     * This method will takes the necessary parameters from parameter list and do the resequencing
     * Resequencing is done through reading messages until the next-to-send message is found
     * If required is not found then waits until the next instance is created.
     *
     * @param jobExecutionContext - a bundle with information related to environment
     * @throws JobExecutionException - to indicate Quartz scheduler that an error occurred while executing the job
     */
    @Override
    public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {

        final JobDataMap jdm = jobExecutionContext.getMergedJobDataMap();
        final MessageStore messageStore = (MessageStore) jdm.get(MessageProcessorConstants.MESSAGE_STORE);
        final ResequencingProcessor processor = (ResequencingProcessor) jdm.get(
                ScheduledMessageProcessor.PROCESSOR_INSTANCE);

        final Map<String, Object> parameters = (Map<String, Object>) jdm.get(MessageProcessorConstants.PARAMETERS);
        final String sequence = (String) parameters.get(ResequencingProcessor.NEXT_SEQUENCE);

        SynapseXPath seqNoxPath = null;

        /** Checking for activation of processor or existence of message store  */
        if (!processor.isActive() || messageStore == null) {
            return;
        }

        /** Extract the SynapseXpath object from parameters to identify the sequence number of the message */
        if (parameters.get(ResequencingProcessor.SEQUENCE_NUMBER_XPATH) != null) {
            seqNoxPath = (SynapseXPath) parameters.get(ResequencingProcessor.SEQUENCE_NUMBER_XPATH);
        }

        /** Extract the number of messages interested to come */
        if (parameters.get(ResequencingProcessor.REQ_INIT_MSGS) != null) {
            processor.setRequiredInitMsgs(new AtomicInteger(Integer.parseInt((String) parameters.get(
                    ResequencingProcessor.REQ_INIT_MSGS))));
        }
        /** Extract the delay wait until the interested messages come */
        if (parameters.get(ResequencingProcessor.REQ_INIT_MSGS_DELAY) != null) {
            processor.setRequiredInitMsgsDelay(new AtomicInteger(Integer.parseInt((String) parameters.get(
                    ResequencingProcessor.REQ_INIT_MSGS_DELAY))));
        }

        /** Extract whether to delete duplicate messages */
        if (parameters.get(ResequencingProcessor.DELETE_DUPLICATES) != null) {
            String result=(String) parameters.get(ResequencingProcessor.DELETE_DUPLICATES);
            if(result.equalsIgnoreCase("TRUE")){
                processor.setDeleteDuplicates(new AtomicBoolean(true));
            }
        }


        if (!processor.isInitSeqNo().get()) {
            /** Deactivating Resequencing processor to avoid executing multiple Job instances */
            processor.deactivate();

            do {
                delay(ResequencingProcessor.STARTING_NUMBER_INIT_DELAY);

                if (messageStore.size() >= processor.getRequiredInitMsgs().get()) {
                    selectStartingSeqNo(processor, messageStore, seqNoxPath);
                    break;
                }
                processor.tried++;
            } while (processor.tried < processor.getRequiredInitMsgsDelay().get());

        }

        if (!processor.isInitSeqNo().get()) {

            while (true) {

                if (messageStore.size() > 0) {
                    selectStartingSeqNo(processor, messageStore, seqNoxPath);

                    if (!processor.isInitSeqNo().get()) {
                        log.warn("Resequencer failed to select starting sequence number with in given timeout !");
                    }

                    break;
                }

            }

        }

        /** Continue to this section happens only after initializing the starting sequence number */
        boolean errorStop = false;
        while (!errorStop) {

            /** Iterate through message store */
            for (int messageIndex = 0; ; messageIndex++) {
                MessageContext messageContext = messageStore.get(messageIndex);

                if (messageContext == null) {
                    errorStop = true;
                    break;
                }

                /** Extract the sequence number from the message */
                int sequenceNo;
                try {
                    sequenceNo = Integer.parseInt(seqNoxPath.stringValueOf(messageContext));
                } catch (Exception e) {
                    log.warn("Can't Find sequence number from message " + e.getMessage());
                    continue;
                }

                String messageId = messageContext.getMessageID();

                /** Remove messages which have less sequence number than required */
                if(sequenceNo<processor.getNextSeqNo() && processor.getDeleteDuplicates()){
                    messageStore.remove(messageId);
                }

                /** Compare the next-to-go sequence number with current message sequence number */
                if (sequenceNo == processor.getNextSeqNo()) {

                    /** Remove selected message from store */

                    messageStore.remove(messageId);
                    /** If sending does not failed increase sequence number */
                    if (send(messageContext, sequence)) {

                        processor.incrementNextSeqNo();
                    }
                    /** Break and start searching from beginning */
                    break;
                }


            }

        }

        /** Reactivating Processor after selecting initial sequence number */
        if (!processor.isActive()) {
            processor.activate();
        }

    }

    /**
     * Selects the smallest sequence number as the starting sequence number from a given message store
     *
     * @param processor    - Resequencing processor which is interested to know starting sequence number
     * @param messageStore - Message store that contains messages
     * @param seqNoxPath   - SynapseXpath object which contains the xpath to find the sequence number from a message
     */
    private void selectStartingSeqNo(ResequencingProcessor processor, MessageStore messageStore,
                                     SynapseXPath seqNoxPath) {
        /** Iterate through message store */
        for (int messageIndex = 0; ; messageIndex++) {
            try {
                MessageContext messageContext = messageStore.get(messageIndex);
                if (messageContext == null) {
                    break;
                }
                /** Extract the sequence number from the message */
                int sequenceNo;

                sequenceNo = Integer.parseInt(seqNoxPath.stringValueOf(messageContext));


                /** If the sequence number is smaller that current next-sequence number, current next-sequence number get replaced */
                if (sequenceNo < processor.getNextSeqNo()) {
                    processor.setNextSeqNo(sequenceNo);
                    processor.setInitSeqNo(new AtomicBoolean(true));
                }


            } catch (NumberFormatException e) {
                handleException("Invalid xPath parameter - Sequence number specified is not an integer ");
            } catch (Exception e) {
                handleException("Failed to initialize starting sequence number at startup: " + e.getMessage());
            }
        }


    }

    /**
     * To timePeriod the processor until next checking up
     * This method is in use wen initializing the starting sequence number of the resequencer
     *
     * @param timePeriod - the time period which waits before a single cycle
     */
    private void delay(long timePeriod) {
        try {
            Thread.sleep(timePeriod);
        } catch (InterruptedException e) {
            log.error("Interrupted while thread sleeping in resequencer", e);
        }
    }

    /**
     * Transmit the message in to a given sequence
     * This method will takes the sequence given in sequence parameter. If no sequence is given this will return false     *
     *
     * @param messageContext - the content of the message that is transferred by Resequencer from message store
     * @param sequence       - the sequence name that the message should be passed
     * @return boolean         - to indicate the success of transferring the message
     */
    private boolean send(MessageContext messageContext, String sequence) {

        Mediator processingSequence = messageContext.getSequence(sequence);
        if (processingSequence != null) {
            processingSequence.mediate(messageContext);
            return true;
        }
        return false;
    }

    /**
     * Handling errors are done here.
     * This will log the error messages and throws SynapseException
     *
     * @param msg - Error message to be set
     * @throws SynapseException - Exception related to Synapse at Runtime
     */
    private static void handleException(String msg) {
        log.error(msg);
        throw new SynapseException(msg);
    }

}
