/**
 * JaDOrT: JASMINe Deployment Orchestration Tool
 * Copyright (C) 2008-2009 Bull S.A.S.
 * Copyright (C) 2008-2009 France Telecom R&D
 * Contact: jasmine@ow2.org
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA
 *
 * --------------------------------------------------------------------------
 * $Id: JMSReceiver.java 5849 2010-01-05 14:50:34Z alitokmen $
 * --------------------------------------------------------------------------
 */
package org.ow2.jasmine.jadort.service.implementation;

import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.OutputStream;
import java.io.PrintStream;
import java.io.StringReader;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Enumeration;
import java.util.logging.Level;
import java.util.logging.Logger;

import javax.annotation.Resource;
import javax.ejb.ActivationConfigProperty;
import javax.ejb.EJBContext;
import javax.ejb.MessageDriven;
import javax.ejb.TransactionAttribute;
import javax.ejb.TransactionAttributeType;
import javax.ejb.TransactionManagement;
import javax.ejb.TransactionManagementType;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.persistence.EntityManager;
import javax.persistence.PersistenceContext;
import javax.transaction.UserTransaction;

import org.ow2.jasmine.jadort.api.JadortServiceException;
import org.ow2.jasmine.jadort.api.entities.deployment.ApplicationBean;
import org.ow2.jasmine.jadort.api.entities.deployment.ProgressBean;
import org.ow2.jasmine.jadort.api.entities.deployment.ProgressState;
import org.ow2.jasmine.jadort.api.entities.deployment.ServerProgressBean;
import org.ow2.jasmine.jadort.api.entities.deployment.ServerProgressState;
import org.ow2.jasmine.jadort.api.entities.deployment.WorkerProgressBean;
import org.ow2.jasmine.jadort.api.entities.deployment.WorkerProgressState;
import org.ow2.jasmine.jadort.api.entities.deployment.ApplicationBean.ApplicationType;
import org.ow2.jasmine.jadort.api.entities.deployment.OperationStateBean.ActionState;
import org.ow2.jasmine.jadort.api.entities.topology.VMBean;
import org.ow2.jasmine.jadort.service.action.AbstractAction;
import org.ow2.jasmine.jadort.service.action.ServerAction;
import org.ow2.jasmine.jadort.service.action.VMMAction;
import org.ow2.jasmine.jadort.service.action.WorkerAction;
import org.ow2.jasmine.vmm.api.VirtualMachineMXBean.PowerState;
import org.ow2.jonas.lib.security.context.Marshalling;
import org.ow2.jonas.lib.security.context.SecurityContext;
import org.ow2.jonas.lib.security.context.SecurityCurrent;

/**
 * Message Driven Bean that allows asynchronous execution of tasks in the
 * maintenance / migration execution step.<br/>
 * <br/>
 * <b>NOTE</b>: This message-driven bean uses bean-managed transactions, since
 * it may use (very) long-running transactions.
 * 
 * @author Malek Chahine
 * @author Remy Bresson
 * @author S. Ali Tokmen
 */
@MessageDriven(activationConfig = {
    @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
    @ActivationConfigProperty(propertyName = "destination", propertyValue = JMSReceiver.QUEUE_NAME)})
@TransactionManagement(TransactionManagementType.BEAN)
public class JMSReceiver implements MessageListener {

    // Constants shared within the package (mostly JMSSender)
    final static String QUEUE_NAME = "queue/JadortServiceJMSQueue";

    final static String SECURITY_CONTEXT = "securityContext";

    final static String SECURITY_SIGNATURE = "securitySignature";

    final static String SERVER_PROGRESS_ID = "serverProgressId";

    final static String USE_TARGET = "useTarget";

    final static String WORKER_PROGRESS_ID = "workerProgressId";

    final static String AIMED_PROGRESS_PERCENT = "aimedProgressPercent";

    final static String AIMED_STATE = "aimedState";

    final static String APPLICATION_ID = "applicationId";

    final static String VM_IMAGE_UUID = "vmImageUUID";

    final static String ACTION_TYPE = "actionType";

    final static String APPLICATION_NAME = "applicationName";

    final static String VM_NAME = "vmName";

    final static String CHECK_IF_HOST_HALTED = "checkIfHostHalted";

    final static String OPERATION_TYPE = "operationType";

    private static final int START_TIMEOUT = 90;

    private static final int STOP_TIMEOUT = 45;

    /**
     * Action type shared within the package (mostly JMSSender)
     */
    enum ActionType {
        UPLOAD,
        DEPLOY,
        SET_DEFAULT,
        UNDEPLOY,
        ERASE,
        START_SERVER,
        STOP_SERVER,
        START_WORKER,
        STOP_WORKER,
        DISABLE_APPLICATIONS,
        ENABLE_APPLICATIONS,
        START_VM,
        CREATE_VM,
        DESTROY_VM,
        STOP_VM
    }

    /**
     * Format used to suffix VM host names in order to keep them unique.
     */
    private static final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMddHHmm");

    /**
     * Logger.
     */
    private Logger logger = Logger.getLogger(this.getClass().getName());

    // keep this public for tests
    @PersistenceContext
    public EntityManager em;

    // keep this public for tests
    @Resource
    public EJBContext context;

    @SuppressWarnings("unchecked")
    @TransactionAttribute(TransactionAttributeType.SUPPORTS)
    public void onMessage(final Message message) {
        try {
            try {
                SecurityCurrent securityCurrent = SecurityCurrent.getCurrent();
                if (securityCurrent != null) {
                    String context = message.getStringProperty(JMSReceiver.SECURITY_CONTEXT);
                    String signature = message.getStringProperty(JMSReceiver.SECURITY_SIGNATURE);
                    if (context != null) {
                        SecurityContext securityContext = Marshalling.unmarshallSecurityContext(context.getBytes());
                        if (signature != null) {
                            securityContext.setSignature(signature.getBytes());
                        }
                        securityCurrent.setSecurityContext(securityContext);
                    }
                }
            } catch (Throwable ignored) {
                // We're not on an OW2 security manager, ignore
            }

            AbstractAction action = null;
            ProgressBean progressBean = null;
            OutputStream stackTrace = null;
            ProgressState aimedState = null;
            Integer aimedProgressPercent = message.getIntProperty(JMSReceiver.AIMED_PROGRESS_PERCENT);
            ActionType actionType = ActionType.valueOf(message.getStringProperty(JMSReceiver.ACTION_TYPE));

            try {
                Enumeration<?> properties = message.getPropertyNames();
                while (properties.hasMoreElements()) {
                    String propertyName = properties.nextElement().toString();
                    if (JMSReceiver.SERVER_PROGRESS_ID.equals(propertyName)) {
                        aimedState = ServerProgressState.valueOf(message.getStringProperty(JMSReceiver.AIMED_STATE));
                        Integer serverProgressId = message.getIntProperty(JMSReceiver.SERVER_PROGRESS_ID);
                        ServerProgressBean serverProgressBean = this.em.find(ServerProgressBean.class, serverProgressId);
                        serverProgressBean.setIsOnTarget(false);
                        progressBean = serverProgressBean;

                        if (ActionType.START_VM.equals(actionType) || ActionType.STOP_VM.equals(actionType)
                            || ActionType.CREATE_VM.equals(actionType) || ActionType.DESTROY_VM.equals(actionType)) {

                            // Before stopping a VM, stop the associated server
                            if (ActionType.STOP_VM.equals(actionType)) {
                                ServerAction serverAction;
                                boolean useTarget = message.getBooleanProperty(JMSReceiver.USE_TARGET);
                                if (useTarget && serverProgressBean.getServer().getTarget() != null) {
                                    serverAction = ServerAction.getServerAction(serverProgressBean.getServer().getTarget());
                                    serverProgressBean.setIsOnTarget(true);
                                } else {
                                    serverAction = ServerAction.getServerAction(serverProgressBean.getServer());
                                }
                                action = serverAction;

                                this.handleServerAction(serverProgressBean, serverAction, ActionType.STOP_SERVER, message);

                                // Append logs which will otherwise get lost
                                progressBean.appendToLog(action.flushLog());
                            }

                            String vmName = message.getStringProperty(JMSReceiver.VM_NAME);
                            VMBean vm = new VMBean();
                            vm.setConnector(serverProgressBean.getServer().getVm().getConnector());
                            VMMAction vmmAction = VMMAction.getVMMAction(vm);
                            action = vmmAction;

                            this.handleVMMAction(serverProgressBean, vmmAction, vmName, actionType, message);

                            // After having started a VM, also start the
                            // associated server
                            if (ActionType.START_VM.equals(actionType)) {
                                ServerAction serverAction;
                                boolean useTarget = message.getBooleanProperty(JMSReceiver.USE_TARGET);
                                if (useTarget && serverProgressBean.getServer().getTarget() != null) {
                                    serverAction = ServerAction.getServerAction(serverProgressBean.getServer().getTarget());
                                    serverProgressBean.setIsOnTarget(true);
                                } else {
                                    serverAction = ServerAction.getServerAction(serverProgressBean.getServer());
                                }

                                // Start the server if necessary
                                if (!serverAction.isStarted()) {
                                    // Don't lose the old logs
                                    progressBean.appendToLog(action.flushLog());

                                    action = serverAction;

                                    this.handleServerAction(serverProgressBean, serverAction, ActionType.START_SERVER, message);
                                }
                            }
                        } else {
                            ServerAction serverAction;
                            if (ActionType.START_SERVER.equals(actionType) || ActionType.STOP_SERVER.equals(actionType)) {
                                boolean useTarget = message.getBooleanProperty(JMSReceiver.USE_TARGET);
                                if (useTarget && serverProgressBean.getServer().getTarget() != null) {
                                    serverAction = ServerAction.getServerAction(serverProgressBean.getServer().getTarget());
                                    serverProgressBean.setIsOnTarget(true);
                                } else {
                                    serverAction = ServerAction.getServerAction(serverProgressBean.getServer());
                                }
                            } else {
                                serverAction = ServerAction.getServerAction(serverProgressBean.getServer());
                            }
                            action = serverAction;

                            this.handleServerAction(serverProgressBean, serverAction, actionType, message);
                        }

                        break;
                    } else if (JMSReceiver.WORKER_PROGRESS_ID.equals(propertyName)) {
                        aimedState = WorkerProgressState.valueOf(message.getStringProperty(JMSReceiver.AIMED_STATE));
                        Integer workerProgressId = message.getIntProperty(JMSReceiver.WORKER_PROGRESS_ID);
                        WorkerProgressBean workerProgressBean = this.em.find(WorkerProgressBean.class, workerProgressId);
                        progressBean = workerProgressBean;

                        WorkerAction workerAction = WorkerAction.getWorkerAction(workerProgressBean.getWorker());
                        action = workerAction;

                        this.handleWorkerAction(workerProgressBean, workerAction, actionType, message);
                        break;
                    }
                }
            } catch (Exception e) {
                stackTrace = new ByteArrayOutputStream();
                e.printStackTrace(new PrintStream(stackTrace));
            }

            if (progressBean != null) {
                if (action != null) {
                    progressBean.appendToLog(action.flushLog());
                }
                if (stackTrace != null) {
                    BufferedReader br = new BufferedReader(new StringReader(stackTrace.toString()));
                    String line;
                    while ((line = br.readLine()) != null) {
                        if (line.length() > 0) {
                            progressBean.appendToLog(line);
                        }
                    }
                }

                progressBean.setProgress(aimedProgressPercent);

                if (stackTrace == null && ActionState.RUNNING.equals(progressBean.getActionState())) {
                    progressBean.setActionState(ActionState.WAITING);
                    progressBean.setProgressState(aimedState);
                } else {
                    progressBean.setActionState(ActionState.FINISHED_ERROR);
                }

                // Create a transaction now, the long-running actions are over
                UserTransaction ut = this.context.getUserTransaction();
                ut.begin();

                this.em.merge(progressBean);
                this.em.flush();

                ut.commit();
            }
        } catch (Throwable t) {
            final String errorMessage = "Failed processing JMS message '" + message + "': " + t.getMessage();
            IllegalStateException exception = new IllegalStateException(errorMessage, t);

            this.logger.log(Level.SEVERE, errorMessage, t);
            throw exception;
        }
    }

    private void handleServerAction(final ServerProgressBean serverProgressBean, final ServerAction serverAction,
        final ActionType actionType, final Message message) throws Exception {
        String applicationName;

        switch (actionType) {
            case UPLOAD:
                Integer applicationId = message.getIntProperty(JMSReceiver.APPLICATION_ID);
                ApplicationBean application = this.em.find(ApplicationBean.class, applicationId);

                serverProgressBean.setOldDeploymentItem(null);

                if (ApplicationType.OSGI.equals(application.getApplicationType())) {
                    for (ApplicationBean applicationOnServer : serverAction.listOfApplications()) {
                        // We need to use startsWith and contains as the version
                        // resolution doesn't work very well on bundles: they
                        // actually seem to be "Present" on JOnAS, not
                        // "Deployed".
                        if (applicationOnServer.getName().startsWith(application.getName())) {
                            if (applicationOnServer.getVersion() != null) {
                                if (applicationOnServer.getVersion().equals(application.getVersion())) {
                                    // This is the same application
                                    break;
                                }
                            } else {
                                // Check for both suffixes: since the
                                // application is "Present", its information is
                                // not set correctly. In particular, its
                                // applicationName contains the file extension
                                final String version = application.getVersion();
                                final String alternativeVersion = version + ".jar";
                                if (applicationOnServer.getName().endsWith(version)
                                    || applicationOnServer.getName().endsWith(alternativeVersion)) {
                                    // This is the same application
                                    break;
                                }
                            }

                            if (serverProgressBean.getOldDeploymentItem() != null) {
                                throw new IllegalStateException("The OSGi deployable \"" + applicationOnServer.getName()
                                    + "\" already has one than one version present on server!");
                            }
                            if (applicationOnServer.getFile() == null) {
                                throw new IllegalStateException("Cannot get the file for the ApplicationBean "
                                    + applicationOnServer);
                            }
                            String file = applicationOnServer.getFile().toString().replace('\\', '/');
                            if (!file.startsWith("/")) {
                                file = '/' + file;
                            }
                            serverProgressBean.setOldDeploymentItem(file);
                        }
                    }
                }

                String newApplication = serverAction.upload(application);
                serverProgressBean.setNewDeploymentItem(newApplication);

                return;

            case DEPLOY:
                applicationName = message.getStringProperty(JMSReceiver.APPLICATION_NAME);

                serverAction.deploy(applicationName);
                return;

            case SET_DEFAULT:
                applicationName = message.getStringProperty(JMSReceiver.APPLICATION_NAME);

                String previousDefault = serverAction.setDefault(applicationName);
                if (previousDefault != null && !previousDefault.endsWith(applicationName)
                    && serverProgressBean.getOldDeploymentItem() == null) {
                    serverProgressBean.setOldDeploymentItem(previousDefault);
                }
                return;

            case UNDEPLOY:
                applicationName = message.getStringProperty(JMSReceiver.APPLICATION_NAME);

                serverAction.undeploy(applicationName);
                return;

            case ERASE:
                applicationName = message.getStringProperty(JMSReceiver.APPLICATION_NAME);

                serverAction.erase(applicationName);
                if (applicationName.equals(serverProgressBean.getNewDeploymentItem())) {
                    // "Undo" of the "upload application" step
                    serverProgressBean.setNewDeploymentItem(null);
                } else if (applicationName.equals(serverProgressBean.getOldDeploymentItem())) {
                    // "Erase old version" step
                    //
                    // Don't do anything, since the canGoToPreviousStep step
                    // requires the oldDeploymentItem in order to determine if
                    // the call to previous() is allowed or not
                }
                return;

            case START_SERVER:
                Exception startException = null;
                try {
                    serverAction.start();
                } catch (Exception e) {
                    startException = e;
                }

                // Start will fail if the server or the manager JMX connector is
                // not present. We might be waiting for the OS to start and the
                // server's JMX connector to become online.
                //
                // As a result, check and if JMX connector offline wait a bit
                try {
                    serverAction.checkJMXConnection();
                } catch (Exception e) {
                    if (serverProgressBean.getServer().getManagerConnector() == null) {
                        // Server not ready yet, ignore the JMX error
                        startException = null;

                        // Wait for the OS to start
                        serverAction.appendToLog("Waiting for server core to start");

                        boolean started = false;
                        final long stopTime = System.currentTimeMillis() + JMSReceiver.START_TIMEOUT * 1000;
                        while (!started && System.currentTimeMillis() < stopTime) {
                            Thread.sleep(1000);
                            try {
                                serverAction.checkJMXConnection();
                                started = true;
                            } catch (Exception ignored) {
                                // If exception, the value of started is not
                                // changed
                            }
                        }

                        if (started) {
                            serverAction.appendToLog("Server core is online");
                        } else {
                            throw new JadortServiceException("Server core did not start after " + JMSReceiver.START_TIMEOUT
                                + " seconds! Try retrying manually using the \"Retry server\" button...", startException);
                        }
                    }
                }

                this.waitForServerState(serverAction, true, startException);
                return;

            case STOP_SERVER:
                Exception stopException = null;
                try {
                    serverAction.stop();
                } catch (Exception e) {
                    stopException = e;
                }

                this.waitForServerState(serverAction, false, stopException);
                serverAction.disconnectJMX();
                return;

            case DISABLE_APPLICATIONS:
                if (!serverAction.enableOrDisableApplications(false)) {
                    serverProgressBean.setActionState(ActionState.FINISHED_ERROR);
                }
                return;

            case ENABLE_APPLICATIONS:
                if (!serverAction.enableOrDisableApplications(true)) {
                    serverProgressBean.setActionState(ActionState.FINISHED_ERROR);
                }
                return;

            default:
                throw new IllegalStateException("Unknown ActionType for a server: " + actionType);
        }
    }

    private void handleWorkerAction(final WorkerProgressBean workerProgressBean, final WorkerAction workerAction,
        final ActionType actionType, final Message message) throws Exception {
        switch (actionType) {
            case START_WORKER:
                workerAction.enable();
                return;

            case STOP_WORKER:
                workerAction.disable();
                return;

            default:
                throw new IllegalStateException("Unknown ActionType for a worker: " + actionType);
        }
    }

    private void handleVMMAction(final ServerProgressBean serverProgressBean, final VMMAction vmmAction, final String vmName,
        final ActionType actionType, final Message message) throws Exception {
        switch (actionType) {
            case START_VM:
                // If VM already running, don't attempt to start it again
                if (!PowerState.RUNNING.equals(vmmAction.getVMState(vmName))) {
                    vmmAction.startVM(vmName);
                }

                // Always check VM state
                vmmAction.waitForVMState(vmName, PowerState.RUNNING, JMSReceiver.START_TIMEOUT);

                return;

            case STOP_VM:
                // If VM already halted, don't attempt to stop it again
                if (!PowerState.HALTED.equals(vmmAction.getVMState(vmName))) {
                    vmmAction.stopVM(vmName);
                }

                // Always check VM state
                vmmAction.waitForVMState(vmName, PowerState.HALTED, JMSReceiver.STOP_TIMEOUT);

                return;

            case CREATE_VM:
                Date now = new Date();

                String vmImageUUID = message.getStringProperty(JMSReceiver.VM_IMAGE_UUID);
                String vmPath = vmName.substring(0, vmName.lastIndexOf('/'));
                String vmNameLabel = vmName.substring(vmName.lastIndexOf('/') + 1);
                if (vmNameLabel.indexOf('_') > 0) {
                    vmNameLabel = vmNameLabel.substring(0, vmNameLabel.indexOf('_'));
                }
                String newVmName = vmPath + '/' + vmNameLabel + '_' + JMSReceiver.dateFormat.format(now);
                vmmAction.deployImageOnVM(vmName, newVmName, vmImageUUID);
                serverProgressBean.setNewDeploymentItem(newVmName);
                return;

            case DESTROY_VM:
                boolean checkIfHostHalted = message.getBooleanProperty(JMSReceiver.CHECK_IF_HOST_HALTED);

                if (checkIfHostHalted) {
                    PowerState currentState = vmmAction.getVMState(vmName);
                    if (!PowerState.HALTED.equals(currentState)) {
                        throw new IllegalStateException("The state of the VM host \"" + vmName + "\" is currently "
                            + currentState + ", it is expected to be " + PowerState.HALTED);
                    }
                }

                vmmAction.destroyVM(vmName);

                if (vmName.equals(serverProgressBean.getNewDeploymentItem())) {
                    // "Undo" of the "maintain server" step
                    serverProgressBean.setNewDeploymentItem(null);
                } else if (vmName.equals(serverProgressBean.getOldDeploymentItem())) {
                    // "Finish" step, do the cleanup
                    serverProgressBean.setOldDeploymentItem(null);
                }
                return;

            default:
                throw new IllegalStateException("Unknown ActionType for a VM: " + actionType);
        }
    }

    private void waitForServerState(final ServerAction serverAction, final boolean state, final Exception potentialCause)
        throws Exception {
        serverAction.appendToLog("Waiting for server to " + (state ? "start" : "stop"));

        final long timeout;
        if (state) {
            timeout = JMSReceiver.START_TIMEOUT;
        } else {
            timeout = JMSReceiver.STOP_TIMEOUT;
        }

        boolean currentState = !state;
        final long stopTime = System.currentTimeMillis() + timeout * 1000;
        while (currentState != state && System.currentTimeMillis() < stopTime) {
            Thread.sleep(1000);
            currentState = serverAction.isStarted();
        }

        if (currentState == state) {
            serverAction.appendToLog("Server is " + (state ? "started" : "stopped"));
        } else {
            throw new JadortServiceException("Server did not " + (state ? "start" : "stop") + " after " + timeout
                + " seconds! Try retrying manually using the \"Retry server\" button...", potentialCause);
        }
    }
}
