<commit_before>package org.zstack.core.scheduler;

import org.quartz.*;
import org.quartz.impl.StdSchedulerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.zstack.core.Platform;
import org.zstack.core.cloudbus.CloudBus;
import org.zstack.core.cloudbus.MessageSafe;
import org.zstack.core.cloudbus.ResourceDestinationMaker;
import org.zstack.core.db.DatabaseFacade;
import org.zstack.core.db.SimpleQuery;
import org.zstack.core.errorcode.ErrorFacade;
import org.zstack.core.thread.AsyncThread;
import org.zstack.header.AbstractService;
import org.zstack.header.apimediator.ApiMessageInterceptionException;
import org.zstack.header.core.scheduler.SchedulerInventory;
import org.zstack.header.core.scheduler.SchedulerStatus;
import org.zstack.header.core.scheduler.SchedulerVO;
import org.zstack.header.core.scheduler.SchedulerVO_;
import org.zstack.header.errorcode.SysErrors;
import org.zstack.header.managementnode.ManagementNodeReadyExtensionPoint;
import org.zstack.header.message.Message;
import org.zstack.utils.Utils;
import org.zstack.utils.gson.JSONObjectUtil;
import org.zstack.utils.logging.CLogger;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

import static org.quartz.CronScheduleBuilder.cronSchedule;
import static org.quartz.JobBuilder.newJob;
import static org.quartz.JobKey.jobKey;
import static org.quartz.SimpleScheduleBuilder.simpleSchedule;
import static org.quartz.TriggerBuilder.newTrigger;
import static org.quartz.TriggerKey.triggerKey;

/**
 * Created by <NAME> on 6/22/16.
 */
public class SchedulerFacadeImpl extends AbstractService implements SchedulerFacade, ManagementNodeReadyExtensionPoint {
    private static final CLogger logger = Utils.getLogger(SchedulerFacadeImpl.class);

    @Autowired
    private transient CloudBus bus;
    @Autowired
    private transient ErrorFacade errf;
    @Autowired
    protected transient DatabaseFacade dbf;
    @Autowired
    private transient ResourceDestinationMaker destinationMaker;

    private Scheduler scheduler;

    protected SchedulerVO self;

    protected SchedulerInventory getInventory() {
        return SchedulerInventory.valueOf(self);
    }
    @Override
    @MessageSafe
    public void handleMessage(Message msg) {
        if (msg instanceof APIDeleteSchedulerMsg) {
            handle((APIDeleteSchedulerMsg) msg);
        } else  if (msg instanceof  APIUpdateSchedulerMsg) {
            handle((APIUpdateSchedulerMsg) msg);
        } else {
            bus.dealWithUnknownMessage(msg);
        }
    }

    private void handle(APIDeleteSchedulerMsg msg) {
        APIDeleteSchedulerEvent evt = new APIDeleteSchedulerEvent(msg.getId());
        SimpleQuery<SchedulerVO> q = dbf.createQuery(SchedulerVO.class);
        q.select(SchedulerVO_.jobName);
        q.add(SchedulerVO_.uuid, SimpleQuery.Op.EQ, msg.getUuid());
        String jobName = q.findValue();
        SimpleQuery<SchedulerVO> q2 = dbf.createQuery(SchedulerVO.class);
        q2.select(SchedulerVO_.jobGroup);
        q2.add(SchedulerVO_.uuid, SimpleQuery.Op.EQ, msg.getUuid());
        String jobGroup = q2.findValue();
        try {
            scheduler.deleteJob(jobKey(jobName, jobGroup));
            dbf.removeByPrimaryKey(msg.getUuid(), SchedulerVO.class);
            bus.publish(evt);
        } catch (SchedulerException e) {
            evt.setErrorCode(errf.instantiateErrorCode(SysErrors.DELETE_RESOURCE_ERROR, e.getMessage()));
            bus.publish(evt);
            logger.warn("Delete Scheduler trigger failed!");
            throw new RuntimeException(e);
        }

    }

    private void handle(APIUpdateSchedulerMsg msg) {
        if (msg.getSchedulerType() == null) {
            throw new ApiMessageInterceptionException(errf.instantiateErrorCode(SysErrors.INVALID_ARGUMENT_ERROR,
                    String.format("schedulerType must be set")
            ));
        }
        SchedulerVO vo = updateScheduler(msg);
        if (vo != null) {
            self = dbf.updateAndRefresh(vo);
        }
        APIUpdateSchedulerEvent evt = new APIUpdateSchedulerEvent(msg.getId());
        evt.setInventory(getInventory());
        bus.publish(evt);
    }

    private SchedulerVO updateScheduler(APIUpdateSchedulerMsg msg) {
        boolean update = false;
        boolean reSchedule = false;
        self = dbf.findByUuid(msg.getUuid(), SchedulerVO.class);
        SimpleQuery<SchedulerVO> q = dbf.createQuery(SchedulerVO.class);
        q.select(SchedulerVO_.triggerName);
        q.add(SchedulerVO_.uuid, SimpleQuery.Op.EQ, msg.getUuid());
        String triggerName = q.findValue();
        SimpleQuery<SchedulerVO> q2 = dbf.createQuery(SchedulerVO.class);
        q2.select(SchedulerVO_.triggerGroup);
        q2.add(SchedulerVO_.uuid, SimpleQuery.Op.EQ, msg.getUuid());
        String triggerGroup = q2.findValue();
        if (msg.getSchedulerName() != null ) {
            self.setSchedulerName(msg.getSchedulerName());
        }
        if (msg.getSchedulerType().equals("simple")) {
            if (msg.getSchedulerInterval() != null ) {
                self.setSchedulerInterval(msg.getSchedulerInterval());
                reSchedule = true;
            }
            if (msg.getRepeatCount() != null ) {
                self.setRepeatCount(msg.getRepeatCount());
                reSchedule = true;
            }
            if ( msg.getStartDate() != null ) {
                self.setStartDate(new Timestamp(msg.getStartDate() * 1000));
                reSchedule = true;
            }
            Trigger oldTrigger = null;
            try {
                oldTrigger = scheduler.getTrigger(triggerKey(triggerName, triggerGroup));
            } catch (SchedulerException e) {
                logger.warn("Get Scheduler trigger failed!");
                throw new RuntimeException(e);
            }
            TriggerBuilder tb = oldTrigger != null ? oldTrigger.getTriggerBuilder() : null;
            Trigger newTrigger = null;
            if (tb != null && reSchedule) {
                if ( msg.getRepeatCount() != 0 ) {
                    newTrigger = tb.withSchedule(simpleSchedule()
                            .withIntervalInSeconds(msg.getSchedulerInterval())
                            .withRepeatCount(msg.getRepeatCount()))
                            .build();
                }
                else {
                    newTrigger = tb.withSchedule(simpleSchedule()
                            .withIntervalInSeconds(msg.getSchedulerInterval())
                            .repeatForever())
                            .build();
                }
            }

            try {
                scheduler.rescheduleJob(oldTrigger != null ? oldTrigger.getKey() : null, newTrigger);
                update = true;
            } catch (SchedulerException e) {
                logger.warn("Reschedule simple Scheduler job failed!");
                throw new RuntimeException(e);
            }

        }
        else if (msg.getSchedulerType().equals("cron")) {
            if ( msg.getCronScheduler() != null ) {
                self.setCronScheduler(msg.getCronScheduler());
            }
            // retrieve the trigger
            Trigger oldTrigger = null;
            try {
                oldTrigger = scheduler.getTrigger(triggerKey(triggerName, triggerGroup));
            } catch (SchedulerException e) {
                logger.warn("Get Scheduler trigger failed!");
                throw new RuntimeException(e);
            }
            TriggerBuilder tb = oldTrigger.getTriggerBuilder();
            Trigger newTrigger = tb.withSchedule(cronSchedule(msg.getCronScheduler()))
                    .build();
            try {
                scheduler.rescheduleJob(oldTrigger.getKey(), newTrigger);
                update = true;
            } catch (SchedulerException e) {
                logger.warn("Reschedule cron Scheduler job failed!");
                throw new RuntimeException(e);
            }
        }

        return update ? self : null;
    }


    public String getId() {

        return bus.makeLocalServiceId(SchedulerConstant.SERVICE_ID);
    }

    private void loadSchedulerJobs() {

        SimpleQuery<SchedulerVO> q = dbf.createQuery(SchedulerVO.class);
        q.select(SchedulerVO_.uuid);
        q.add(SchedulerVO_.managementNodeUuid, SimpleQuery.Op.NULL);
        List<String> ids = q.listValue();

        List<String> ours = new ArrayList<String>();
        for (String id : ids) {
            if (destinationMaker.isManagedByUs(id)) {
                ours.add(id);<commit_msg>Fix some issues on update scheduler API<commit_after>package org.zstack.core.scheduler;

import org.quartz.*;
import org.quartz.impl.StdSchedulerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.zstack.core.Platform;
import org.zstack.core.cloudbus.CloudBus;
import org.zstack.core.cloudbus.MessageSafe;
import org.zstack.core.cloudbus.ResourceDestinationMaker;
import org.zstack.core.db.DatabaseFacade;
import org.zstack.core.db.SimpleQuery;
import org.zstack.core.errorcode.ErrorFacade;
import org.zstack.core.thread.AsyncThread;
import org.zstack.header.AbstractService;
import org.zstack.header.core.scheduler.SchedulerInventory;
import org.zstack.header.core.scheduler.SchedulerStatus;
import org.zstack.header.core.scheduler.SchedulerVO;
import org.zstack.header.core.scheduler.SchedulerVO_;
import org.zstack.header.errorcode.SysErrors;
import org.zstack.header.managementnode.ManagementNodeReadyExtensionPoint;
import org.zstack.header.message.Message;
import org.zstack.utils.Utils;
import org.zstack.utils.gson.JSONObjectUtil;
import org.zstack.utils.logging.CLogger;

import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

import static org.quartz.CronScheduleBuilder.cronSchedule;
import static org.quartz.JobBuilder.newJob;
import static org.quartz.JobKey.jobKey;
import static org.quartz.SimpleScheduleBuilder.simpleSchedule;
import static org.quartz.TriggerBuilder.newTrigger;
import static org.quartz.TriggerKey.triggerKey;

/**
 * Created by <NAME> on 6/22/16.
 */
public class SchedulerFacadeImpl extends AbstractService implements SchedulerFacade, ManagementNodeReadyExtensionPoint {
    private static final CLogger logger = Utils.getLogger(SchedulerFacadeImpl.class);

    @Autowired
    private transient CloudBus bus;
    @Autowired
    private transient ErrorFacade errf;
    @Autowired
    protected transient DatabaseFacade dbf;
    @Autowired
    private transient ResourceDestinationMaker destinationMaker;

    private Scheduler scheduler;

    protected SchedulerVO self;

    protected SchedulerInventory getInventory() {
        return SchedulerInventory.valueOf(self);
    }
    @Override
    @MessageSafe
    public void handleMessage(Message msg) {
        if (msg instanceof APIDeleteSchedulerMsg) {
            handle((APIDeleteSchedulerMsg) msg);
        } else  if (msg instanceof  APIUpdateSchedulerMsg) {
            handle((APIUpdateSchedulerMsg) msg);
        } else {
            bus.dealWithUnknownMessage(msg);
        }
    }

    private void handle(APIDeleteSchedulerMsg msg) {
        APIDeleteSchedulerEvent evt = new APIDeleteSchedulerEvent(msg.getId());
        SimpleQuery<SchedulerVO> q = dbf.createQuery(SchedulerVO.class);
        q.select(SchedulerVO_.jobName);
        q.add(SchedulerVO_.uuid, SimpleQuery.Op.EQ, msg.getUuid());
        String jobName = q.findValue();
        SimpleQuery<SchedulerVO> q2 = dbf.createQuery(SchedulerVO.class);
        q2.select(SchedulerVO_.jobGroup);
        q2.add(SchedulerVO_.uuid, SimpleQuery.Op.EQ, msg.getUuid());
        String jobGroup = q2.findValue();
        try {
            scheduler.deleteJob(jobKey(jobName, jobGroup));
            dbf.removeByPrimaryKey(msg.getUuid(), SchedulerVO.class);
            bus.publish(evt);
        } catch (SchedulerException e) {
            evt.setErrorCode(errf.instantiateErrorCode(SysErrors.DELETE_RESOURCE_ERROR, e.getMessage()));
            bus.publish(evt);
            logger.warn("Delete Scheduler trigger failed!");
            throw new RuntimeException(e);
        }

    }

    private void handle(APIUpdateSchedulerMsg msg) {

        SchedulerVO vo = updateScheduler(msg);
        if (vo != null) {
            self = dbf.updateAndRefresh(vo);
        }
        APIUpdateSchedulerEvent evt = new APIUpdateSchedulerEvent(msg.getId());
        evt.setInventory(getInventory());
        bus.publish(evt);
    }

    private SchedulerVO updateScheduler(APIUpdateSchedulerMsg msg) {
        boolean reSchedule = false;
        self = dbf.findByUuid(msg.getUuid(), SchedulerVO.class);
        SimpleQuery<SchedulerVO> q = dbf.createQuery(SchedulerVO.class);
        q.select(SchedulerVO_.triggerName);
        q.add(SchedulerVO_.uuid, SimpleQuery.Op.EQ, msg.getUuid());
        String triggerName = q.findValue();
        SimpleQuery<SchedulerVO> q2 = dbf.createQuery(SchedulerVO.class);
        q2.select(SchedulerVO_.triggerGroup);
        q2.add(SchedulerVO_.uuid, SimpleQuery.Op.EQ, msg.getUuid());
        String triggerGroup = q2.findValue();
        if ( msg.getSchedulerName() != null ) {
            self.setSchedulerName(msg.getSchedulerName());
        }

        if ( msg.getSchedulerType() != null ) {
            reSchedule = true;
            self.setSchedulerType(msg.getSchedulerType());
        }

        if ( msg.getSchedulerInterval() != null ) {
            reSchedule = true;
            self.setSchedulerInterval(msg.getSchedulerInterval());
        }

        if ( msg.getRepeatCount() != null ) {
            reSchedule = true;
            self.setRepeatCount(msg.getRepeatCount());
        }

        if ( msg.getStartDate() != null ) {
            reSchedule = true;
            self.setStartDate(new Timestamp(msg.getStartDate() * 1000));
        }

        if ( msg.getCronScheduler() != null ) {
            reSchedule = true;
            self.setCronScheduler(msg.getCronScheduler());
        }

        if (self.getSchedulerType().equals("simple") && reSchedule) {
            Trigger oldTrigger = null;
            try {
                oldTrigger = scheduler.getTrigger(triggerKey(triggerName, triggerGroup));
            } catch (SchedulerException e) {
                logger.warn("Get Scheduler trigger failed!");
                throw new RuntimeException(e);
            }
            TriggerBuilder tb =  oldTrigger.getTriggerBuilder();
            Trigger newTrigger = null;
            if (tb != null) {
                if ( self.getRepeatCount() != 0 ) {
                    newTrigger = tb.withSchedule(simpleSchedule()
                            .withIntervalInSeconds(self.getSchedulerInterval())
                            .withRepeatCount(self.getRepeatCount()))
                            .build();
                }
                else {
                    newTrigger = tb.withSchedule(simpleSchedule()
                            .withIntervalInSeconds(self.getSchedulerInterval())
                            .repeatForever())
                            .build();
                }
            }

            try {
                scheduler.rescheduleJob(oldTrigger.getKey(), newTrigger);
            } catch (SchedulerException e) {
                logger.warn("Reschedule simple Scheduler job failed!");
                throw new RuntimeException(e);
            }

        } else if (self.getSchedulerType().equals("cron") && reSchedule) {
            if ( self.getCronScheduler() != null ) {
                self.setCronScheduler(self.getCronScheduler());
            }
            // retrieve the trigger
            Trigger oldTrigger = null;
            try {
                oldTrigger = scheduler.getTrigger(triggerKey(triggerName, triggerGroup));
            } catch (SchedulerException e) {
                logger.warn("Get Scheduler trigger failed!");
                throw new RuntimeException(e);
            }
            TriggerBuilder tb = oldTrigger.getTriggerBuilder();
            Trigger newTrigger = tb.withSchedule(cronSchedule(self.getCronScheduler()))
                    .build();
            try {
                scheduler.rescheduleJob(oldTrigger.getKey(), newTrigger);
            } catch (SchedulerException e) {
                logger.warn("Reschedule cron Scheduler job failed!");
                throw new RuntimeException(e);
            }
        }

        return self;
    }


    public String getId() {

        return bus.makeLocalServiceId(SchedulerConstant.SERVICE_ID);
    }

    private void loadSchedulerJobs() {

        SimpleQuery<SchedulerVO> q = dbf.createQuery(SchedulerVO.class);
        q.select(SchedulerVO_.uuid);
        q.add(SchedulerVO_.managementNodeUuid, SimpleQuery.Op.NULL);
        List<String> ids = q.listValue();

        List<String> ours = new ArrayList<String>();
        for (String id : ids) {
            if (destinationMaker.isManagedByUs(id)) {
                ours.add(id);
{'fraction_non_alphanumeric': 0.10066016241163755, 'fraction_numerical': 0.0021031722848630016, 'mean_word_length': 2.4277132559070886, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6000972', 'n_tokens_mistral': 4766, 'n_tokens_neox': 4680, 'n_words': 843}