diff --git a/mifosng-db/migrations/core_db/V103__cluster_support_for_batch_jobs.sql b/mifosng-db/migrations/core_db/V103__cluster_support_for_batch_jobs.sql new file mode 100755 index 000000000..7343f7f74 --- /dev/null +++ b/mifosng-db/migrations/core_db/V103__cluster_support_for_batch_jobs.sql @@ -0,0 +1,14 @@ +ALTER TABLE `job` + ADD COLUMN `is_misfired` TINYINT(1) NOT NULL DEFAULT '0' AFTER `scheduler_group`; + +CREATE TABLE `scheduler_detail` ( + `id` SMALLINT(2) NOT NULL AUTO_INCREMENT, + `is_suspended` TINYINT(1) NOT NULL DEFAULT '0', + `execute_misfired_jobs` TINYINT(1) NOT NULL DEFAULT '1', + `reset_scheduler_on_bootup` TINYINT(1) NOT NULL DEFAULT '1', + PRIMARY KEY (`id`) +) +COLLATE='utf8_general_ci' +ENGINE=InnoDB; + +INSERT INTO `scheduler_detail` (`is_suspended`, `execute_misfired_jobs`, `reset_scheduler_on_bootup`) VALUES (FALSE,TRUE,TRUE); diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/ScheduledJobDetail.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/ScheduledJobDetail.java index 7664cc5bc..48b6ba191 100644 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/ScheduledJobDetail.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/ScheduledJobDetail.java @@ -28,6 +28,7 @@ public class ScheduledJobDetail extends AbstractPersistable { @Column(name = "cron_expression") private String cronExpression; + @SuppressWarnings("unused") @Column(name = "create_time") @Temporal(TemporalType.TIMESTAMP) private Date createTime; @@ -38,6 +39,7 @@ public class ScheduledJobDetail extends AbstractPersistable { @Column(name = "group_name") private String groupName; + @SuppressWarnings("unused") @Column(name = "previous_run_start_time") @Temporal(TemporalType.TIMESTAMP) private Date previousRunStartTime; @@ -49,6 +51,7 @@ public class ScheduledJobDetail extends AbstractPersistable { @Column(name = "job_key") private String jobKey; + @SuppressWarnings("unused") @Column(name = "initializing_errorlog") private String errorLog; @@ -58,12 +61,16 @@ public class ScheduledJobDetail extends AbstractPersistable { @Column(name = "currently_running") private boolean currentlyRunning; + @SuppressWarnings("unused") @Column(name = "updates_allowed") private boolean updatesAllowed; @Column(name = "scheduler_group") private Short schedulerGroup; + @Column(name = "is_misfired") + private boolean triggerMisfired; + protected ScheduledJobDetail() { } @@ -104,6 +111,10 @@ public class ScheduledJobDetail extends AbstractPersistable { this.previousRunStartTime = previousRunStartTime; } + public Date getNextRunTime() { + return this.nextRunTime; + } + public void updateNextRunTime(final Date nextRunTime) { this.nextRunTime = nextRunTime; } @@ -147,4 +158,12 @@ public class ScheduledJobDetail extends AbstractPersistable { return actualChanges; } + public boolean isTriggerMisfired() { + return this.triggerMisfired; + } + + public void updateTriggerMisfired(boolean triggerMisfired) { + this.triggerMisfired = triggerMisfired; + } + } diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/ScheduledJobDetailRepository.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/ScheduledJobDetailRepository.java index c673f7ea0..50cb0dc5f 100644 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/ScheduledJobDetailRepository.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/ScheduledJobDetailRepository.java @@ -1,7 +1,10 @@ package org.mifosplatform.infrastructure.jobs.domain; +import javax.persistence.LockModeType; + import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.JpaSpecificationExecutor; +import org.springframework.data.jpa.repository.Lock; import org.springframework.data.jpa.repository.Query; import org.springframework.data.repository.query.Param; @@ -14,4 +17,9 @@ public interface ScheduledJobDetailRepository extends JpaRepository { @ManyToOne @JoinColumn(name = "job_id") + @SuppressWarnings("unused") private ScheduledJobDetail scheduledJobDetail; @Column(name = "version") + @SuppressWarnings("unused") private Long version; @Column(name = "start_time") @Temporal(TemporalType.TIMESTAMP) + @SuppressWarnings("unused") private Date startTime; @Column(name = "end_time") @Temporal(TemporalType.TIMESTAMP) + @SuppressWarnings("unused") private Date endTime; @Column(name = "status") + @SuppressWarnings("unused") private String status; @Column(name = "error_message") + @SuppressWarnings("unused") private String errorMessage; @Column(name = "trigger_type") + @SuppressWarnings("unused") private String triggerType; @Column(name = "error_log") + @SuppressWarnings("unused") private String errorLog; public ScheduledJobRunHistory() { diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/SchedulerDetail.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/SchedulerDetail.java new file mode 100755 index 000000000..3cb68eee3 --- /dev/null +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/SchedulerDetail.java @@ -0,0 +1,49 @@ +package org.mifosplatform.infrastructure.jobs.domain; + +import javax.persistence.Column; +import javax.persistence.Entity; +import javax.persistence.Table; + +import org.springframework.data.jpa.domain.AbstractPersistable; + +@Entity +@Table(name = "scheduler_detail") +public class SchedulerDetail extends AbstractPersistable { + + @Column(name = "execute_misfired_jobs") + private boolean executeInstructionForMisfiredJobs; + + @Column(name = "is_suspended") + private boolean suspended; + + @Column(name = "reset_scheduler_on_bootup") + private boolean resetSchedulerOnBootup; + + protected SchedulerDetail() { + + } + + public boolean isExecuteInstructionForMisfiredJobs() { + return this.executeInstructionForMisfiredJobs; + } + + public void updateExecuteInstructionForMisfiredJobs(boolean executeInstructionForMisfiredJobs) { + this.executeInstructionForMisfiredJobs = executeInstructionForMisfiredJobs; + } + + public boolean isSuspended() { + return this.suspended; + } + + public void updateSuspendedState(boolean suspended) { + this.suspended = suspended; + } + + public boolean isResetSchedulerOnBootup() { + return this.resetSchedulerOnBootup; + } + + public void updateResetSchedulerOnBootup(boolean resetSchedulerOnBootup) { + this.resetSchedulerOnBootup = resetSchedulerOnBootup; + } +} diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/SchedulerDetailRepository.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/SchedulerDetailRepository.java new file mode 100755 index 000000000..ee3d8cd8a --- /dev/null +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/SchedulerDetailRepository.java @@ -0,0 +1,8 @@ +package org.mifosplatform.infrastructure.jobs.domain; + +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.JpaSpecificationExecutor; + +public interface SchedulerDetailRepository extends JpaRepository, JpaSpecificationExecutor { + +} diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterServiceImpl.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterServiceImpl.java index c395cfd71..405e8d156 100644 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterServiceImpl.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterServiceImpl.java @@ -4,7 +4,6 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; import java.util.List; -import java.util.Map; import java.util.Properties; import java.util.TimeZone; @@ -15,6 +14,7 @@ import org.mifosplatform.infrastructure.core.exception.PlatformInternalServerExc import org.mifosplatform.infrastructure.core.service.ThreadLocalContextUtil; import org.mifosplatform.infrastructure.jobs.annotation.CronMethodParser; import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobDetail; +import org.mifosplatform.infrastructure.jobs.domain.SchedulerDetail; import org.mifosplatform.infrastructure.jobs.exception.JobNotFoundException; import org.mifosplatform.infrastructure.security.service.TenantDetailsService; import org.quartz.JobDataMap; @@ -24,6 +24,7 @@ import org.quartz.JobListener; import org.quartz.Scheduler; import org.quartz.SchedulerException; import org.quartz.Trigger; +import org.quartz.TriggerListener; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; @@ -50,8 +51,6 @@ public class JobRegisterServiceImpl implements JobRegisterService { private final TenantDetailsService tenantDetailsService; - private final HashMap schedulerStatus = new HashMap(2); - private final HashMap schedulers = new HashMap(4); @Autowired @@ -68,24 +67,32 @@ public class JobRegisterServiceImpl implements JobRegisterService { for (MifosPlatformTenant tenant : allTenants) { ThreadLocalContextUtil.setTenant(tenant); List scheduledJobDetails = this.schedularWritePlatformService.retrieveAllJobs(); - schedulerStatus.put(tenant.getId(), true); for (ScheduledJobDetail jobDetails : scheduledJobDetails) { scheduleJob(jobDetails); + jobDetails.updateTriggerMisfired(false); this.schedularWritePlatformService.saveOrUpdate(jobDetails); } + SchedulerDetail schedulerDetail = this.schedularWritePlatformService.retriveSchedulerDetail(); + if (schedulerDetail.isResetSchedulerOnBootup()) { + schedulerDetail.updateSuspendedState(false); + this.schedularWritePlatformService.updateSchedulerDetail(schedulerDetail); + } } } - public void executeJob(final ScheduledJobDetail scheduledJobDetail) { + public void executeJob(final ScheduledJobDetail scheduledJobDetail, String triggerType) { try { final JobDataMap jobDataMap = new JobDataMap(); - jobDataMap.put(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE, SchedulerServiceConstants.TRIGGER_TYPE_APPLICATION); + if (triggerType == null) { + triggerType = SchedulerServiceConstants.TRIGGER_TYPE_APPLICATION; + } + jobDataMap.put(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE, triggerType); jobDataMap.put(SchedulerServiceConstants.TENANT_IDENTIFIER, ThreadLocalContextUtil.getTenant().getTenantIdentifier()); String key = scheduledJobDetail.getJobKey(); JobKey jobKey = constructJobKey(key); String schedulerName = getSchedulerName(scheduledJobDetail); Scheduler scheduler = schedulers.get(schedulerName); - if (scheduler == null || scheduler.isInStandbyMode() || !scheduler.checkExists(jobKey)) { + if (scheduler == null || !scheduler.checkExists(jobKey)) { JobDetail jobDetail = createJobDetail(scheduledJobDetail); String tempSchedulerName = "temp" + scheduledJobDetail.getId(); List> listenerClasses = new ArrayList>(2); @@ -101,7 +108,6 @@ public class JobRegisterServiceImpl implements JobRegisterService { } } catch (Exception e) { - e.printStackTrace(); throw new PlatformInternalServerException("error.msg.sheduler.job.execution.failed", "Job execution failed for job with id:" + scheduledJobDetail.getId(), scheduledJobDetail.getId()); } @@ -128,32 +134,48 @@ public class JobRegisterServiceImpl implements JobRegisterService { @Override public void pauseScheduler() { - schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), false); - for (Map.Entry schedulerEntry : this.schedulers.entrySet()) { - try { - String schedulerName = schedulerEntry.getValue().getSchedulerName(); - if (isCurrentTenantsScheduler(schedulerName)) { - schedulerEntry.getValue().standby(); - } - } catch (SchedulerException e) { - logger.error(e.getMessage(), e); - } + SchedulerDetail schedulerDetail = this.schedularWritePlatformService.retriveSchedulerDetail(); + if (!schedulerDetail.isSuspended()) { + schedulerDetail.updateSuspendedState(true); + this.schedularWritePlatformService.updateSchedulerDetail(schedulerDetail); } } @Override public void startScheduler() { - for (Map.Entry schedulerEntry : this.schedulers.entrySet()) { - try { - String schedulerName = schedulerEntry.getValue().getSchedulerName(); - if (isCurrentTenantsScheduler(schedulerName)) { - schedulerEntry.getValue().start(); + SchedulerDetail schedulerDetail = this.schedularWritePlatformService.retriveSchedulerDetail(); + if (schedulerDetail.isSuspended()) { + schedulerDetail.updateSuspendedState(false); + this.schedularWritePlatformService.updateSchedulerDetail(schedulerDetail); + if (schedulerDetail.isExecuteInstructionForMisfiredJobs()) { + List scheduledJobDetails = this.schedularWritePlatformService.retrieveAllJobs(); + for (ScheduledJobDetail jobDetail : scheduledJobDetails) { + if (jobDetail.isTriggerMisfired()) { + if (jobDetail.isActiveSchedular()) { + executeJob(jobDetail, SchedulerServiceConstants.TRIGGER_TYPE_CRON); + } + String schedulerName = getSchedulerName(jobDetail); + Scheduler scheduler = schedulers.get(schedulerName); + if (scheduler != null) { + String key = jobDetail.getJobKey(); + JobKey jobKey = constructJobKey(key); + try { + List triggers = scheduler.getTriggersOfJob(jobKey); + for (Trigger trigger : triggers) { + if (trigger.getNextFireTime() != null && trigger.getNextFireTime().after(jobDetail.getNextRunTime())) { + jobDetail.updateNextRunTime(trigger.getNextFireTime()); + } + } + } catch (SchedulerException e) { + logger.error(e.getMessage(), e); + } + } + jobDetail.updateTriggerMisfired(false); + this.schedularWritePlatformService.saveOrUpdate(jobDetail); + } } - } catch (SchedulerException e) { - logger.error(e.getMessage(), e); } } - schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), true); } @Override @@ -166,23 +188,12 @@ public class JobRegisterServiceImpl implements JobRegisterService { public void executeJob(final Long jobId) { ScheduledJobDetail scheduledJobDetail = this.schedularWritePlatformService.findByJobId(jobId); if (scheduledJobDetail == null) { throw new JobNotFoundException(String.valueOf(jobId)); } - executeJob(scheduledJobDetail); + executeJob(scheduledJobDetail, null); } @Override public boolean isSchedulerRunning() { - return this.schedulerStatus.get(ThreadLocalContextUtil.getTenant().getId()); - } - - private boolean isCurrentTenantsScheduler(final String schedulerName) { - int beginIndex = SchedulerServiceConstants.SCHEDULER.length(); - int endIndex = schedulerName.length(); - if (schedulerName.contains(SchedulerServiceConstants.SCHEDULER_GROUP)) { - endIndex = schedulerName.indexOf(SchedulerServiceConstants.SCHEDULER_GROUP); - } - String tenantId = schedulerName.substring(beginIndex, endIndex); - MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant(); - return tenant.getId() == Long.valueOf(tenantId); + return !this.schedularWritePlatformService.retriveSchedulerDetail().isSuspended(); } private void scheduleJob(final ScheduledJobDetail scheduledJobDetails) { @@ -195,9 +206,6 @@ public class JobRegisterServiceImpl implements JobRegisterService { JobDetail jobDetail = createJobDetail(scheduledJobDetails); Trigger trigger = createTrigger(scheduledJobDetails, jobDetail); Scheduler scheduler = getScheduler(scheduledJobDetails); - if (!this.schedulerStatus.get(ThreadLocalContextUtil.getTenant().getId()) && !scheduler.isInStandbyMode()) { - scheduler.standby(); - } scheduler.scheduleJob(jobDetail, trigger); scheduledJobDetails.updateJobKey(getJobKeyAsString(jobDetail.getKey())); scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime()); @@ -253,6 +261,9 @@ public class JobRegisterServiceImpl implements JobRegisterService { listenerClasses.add(SchedulerJobListener.class); } schedulerFactoryBean.setGlobalJobListeners(getGlobalListener(listenerClasses)); + TriggerListener globalTriggerListener = applicationContext.getBean("schedulerTriggerListener", TriggerListener.class); + TriggerListener[] globalTriggerListeners = { globalTriggerListener }; + schedulerFactoryBean.setGlobalTriggerListeners(globalTriggerListeners); Properties quartzProperties = new Properties(); quartzProperties.put(SchedulerFactoryBean.PROP_THREAD_COUNT, Integer.toString(noOfThreads)); schedulerFactoryBean.setQuartzProperties(quartzProperties); diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularWritePlatformService.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularWritePlatformService.java index b5c6a680c..193294ee0 100644 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularWritePlatformService.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularWritePlatformService.java @@ -6,6 +6,7 @@ import org.mifosplatform.infrastructure.core.api.JsonCommand; import org.mifosplatform.infrastructure.core.data.CommandProcessingResult; import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobDetail; import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobRunHistory; +import org.mifosplatform.infrastructure.jobs.domain.SchedulerDetail; public interface SchedularWritePlatformService { @@ -21,6 +22,12 @@ public interface SchedularWritePlatformService { public ScheduledJobDetail findByJobId(Long jobId); - CommandProcessingResult updateJobDetail(Long jobId, JsonCommand command); + public CommandProcessingResult updateJobDetail(Long jobId, JsonCommand command); + + public SchedulerDetail retriveSchedulerDetail(); + + public void updateSchedulerDetail(final SchedulerDetail schedulerDetail); + + public boolean processJobDetailForExecution(String jobKey, String triggerType); } diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularWritePlatformServiceJpaRepositoryImpl.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularWritePlatformServiceJpaRepositoryImpl.java index 74a3b96a9..b686ef849 100644 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularWritePlatformServiceJpaRepositoryImpl.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularWritePlatformServiceJpaRepositoryImpl.java @@ -1,5 +1,6 @@ package org.mifosplatform.infrastructure.jobs.service; +import java.util.Date; import java.util.List; import java.util.Map; @@ -11,6 +12,8 @@ import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobDetail; import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobDetailRepository; import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobRunHistory; import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobRunHistoryRepository; +import org.mifosplatform.infrastructure.jobs.domain.SchedulerDetail; +import org.mifosplatform.infrastructure.jobs.domain.SchedulerDetailRepository; import org.mifosplatform.infrastructure.jobs.exception.JobNotFoundException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; @@ -23,13 +26,17 @@ public class SchedularWritePlatformServiceJpaRepositoryImpl implements Schedular private final ScheduledJobRunHistoryRepository scheduledJobRunHistoryRepository; + private final SchedulerDetailRepository schedulerDetailRepository; + private final JobDetailDataValidator dataValidator; @Autowired public SchedularWritePlatformServiceJpaRepositoryImpl(final ScheduledJobDetailRepository scheduledJobDetailsRepository, - final ScheduledJobRunHistoryRepository scheduledJobRunHistoryRepository, final JobDetailDataValidator dataValidator) { + final ScheduledJobRunHistoryRepository scheduledJobRunHistoryRepository, final JobDetailDataValidator dataValidator, + final SchedulerDetailRepository schedulerDetailRepository) { this.scheduledJobDetailsRepository = scheduledJobDetailsRepository; this.scheduledJobRunHistoryRepository = scheduledJobRunHistoryRepository; + this.schedulerDetailRepository = schedulerDetailRepository; this.dataValidator = dataValidator; } @@ -71,6 +78,22 @@ public class SchedularWritePlatformServiceJpaRepositoryImpl implements Schedular return this.scheduledJobDetailsRepository.findByJobId(jobId); } + @Override + @Transactional + public void updateSchedulerDetail(final SchedulerDetail schedulerDetail) { + this.schedulerDetailRepository.save(schedulerDetail); + } + + @Override + public SchedulerDetail retriveSchedulerDetail() { + SchedulerDetail schedulerDetail = null; + List schedulerDetailList = schedulerDetailRepository.findAll(); + if (schedulerDetailList != null) { + schedulerDetail = schedulerDetailList.get(0); + } + return schedulerDetail; + } + @Transactional @Override public CommandProcessingResult updateJobDetail(final Long jobId, final JsonCommand command) { @@ -89,4 +112,24 @@ public class SchedularWritePlatformServiceJpaRepositoryImpl implements Schedular } + @Transactional + @Override + public boolean processJobDetailForExecution(String jobKey, String triggerType) { + boolean isStopExecution = false; + final ScheduledJobDetail scheduledJobDetail = scheduledJobDetailsRepository.findByJobKeyWithLock(jobKey); + if (scheduledJobDetail.isCurrentlyRunning() + || (triggerType == SchedulerServiceConstants.TRIGGER_TYPE_CRON && (scheduledJobDetail.getNextRunTime().after(new Date())))) { + isStopExecution = true; + } + final SchedulerDetail schedulerDetail = retriveSchedulerDetail(); + if (triggerType == SchedulerServiceConstants.TRIGGER_TYPE_CRON && schedulerDetail.isSuspended()) { + scheduledJobDetail.updateTriggerMisfired(true); + isStopExecution = true; + } else if (!isStopExecution) { + scheduledJobDetail.updateCurrentlyRunningStatus(true); + } + scheduledJobDetailsRepository.save(scheduledJobDetail); + return isStopExecution; + } + } diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerJobListener.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerJobListener.java index a83811f07..542397498 100644 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerJobListener.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerJobListener.java @@ -2,12 +2,9 @@ package org.mifosplatform.infrastructure.jobs.service; import java.util.Date; -import org.mifosplatform.infrastructure.core.domain.MifosPlatformTenant; import org.mifosplatform.infrastructure.core.service.ThreadLocalContextUtil; import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobDetail; import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobRunHistory; -import org.mifosplatform.infrastructure.jobs.exception.JobInProcessExecution; -import org.mifosplatform.infrastructure.security.service.TenantDetailsService; import org.quartz.JobExecutionContext; import org.quartz.JobExecutionException; import org.quartz.JobKey; @@ -31,12 +28,9 @@ public class SchedulerJobListener implements JobListener { private final SchedularWritePlatformService schedularService; - private final TenantDetailsService tenantDetailsService; - @Autowired - public SchedulerJobListener(final SchedularWritePlatformService schedularService, final TenantDetailsService tenantDetailsService) { + public SchedulerJobListener(final SchedularWritePlatformService schedularService) { this.schedularService = schedularService; - this.tenantDetailsService = tenantDetailsService; } @Override @@ -45,16 +39,7 @@ public class SchedulerJobListener implements JobListener { } @Override - public void jobToBeExecuted(final JobExecutionContext context) { - String tenantIdentifier = context.getTrigger().getJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER); - MifosPlatformTenant tenant = this.tenantDetailsService.loadTenantById(tenantIdentifier); - ThreadLocalContextUtil.setTenant(tenant); - JobKey key = context.getTrigger().getJobKey(); - String jobKey = key.getName() + SchedulerServiceConstants.JOB_KEY_SEPERATOR + key.getGroup(); - final ScheduledJobDetail scheduledJobDetail = schedularService.findByJobKey(jobKey); - if (scheduledJobDetail.isCurrentlyRunning()) { throw new JobInProcessExecution(scheduledJobDetail.getJobName()); } - scheduledJobDetail.updateCurrentlyRunningStatus(true); - this.schedularService.saveOrUpdate(scheduledJobDetail); + public void jobToBeExecuted(@SuppressWarnings("unused") final JobExecutionContext context) { } @@ -93,7 +78,8 @@ public class SchedulerJobListener implements JobListener { if (context.getMergedJobDataMap().containsKey(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE)) { triggerType = context.getMergedJobDataMap().getString(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE); } - if (triggerType == SchedulerServiceConstants.TRIGGER_TYPE_CRON) { + if (triggerType == SchedulerServiceConstants.TRIGGER_TYPE_CRON && trigger.getNextFireTime() != null + && trigger.getNextFireTime().after(scheduledJobDetails.getNextRunTime())) { scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime()); } diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerTriggerListener.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerTriggerListener.java new file mode 100755 index 000000000..b33daeeae --- /dev/null +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerTriggerListener.java @@ -0,0 +1,66 @@ +package org.mifosplatform.infrastructure.jobs.service; + +import org.mifosplatform.infrastructure.core.domain.MifosPlatformTenant; +import org.mifosplatform.infrastructure.core.service.ThreadLocalContextUtil; +import org.mifosplatform.infrastructure.security.service.TenantDetailsService; +import org.quartz.JobExecutionContext; +import org.quartz.JobKey; +import org.quartz.Trigger; +import org.quartz.Trigger.CompletedExecutionInstruction; +import org.quartz.TriggerListener; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +@Component +public class SchedulerTriggerListener implements TriggerListener { + + private final String name = "Global trigger Listner"; + + private final SchedularWritePlatformService schedularService; + + private final TenantDetailsService tenantDetailsService; + + @Autowired + public SchedulerTriggerListener(final SchedularWritePlatformService schedularService, final TenantDetailsService tenantDetailsService) { + this.schedularService = schedularService; + this.tenantDetailsService = tenantDetailsService; + + } + + @Override + public String getName() { + return this.name; + } + + @Override + public void triggerFired(@SuppressWarnings("unused") Trigger trigger, @SuppressWarnings("unused") JobExecutionContext context) { + + } + + @Override + public boolean vetoJobExecution(Trigger trigger, JobExecutionContext context) { + + String tenantIdentifier = trigger.getJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER); + MifosPlatformTenant tenant = this.tenantDetailsService.loadTenantById(tenantIdentifier); + ThreadLocalContextUtil.setTenant(tenant); + JobKey key = trigger.getJobKey(); + String jobKey = key.getName() + SchedulerServiceConstants.JOB_KEY_SEPERATOR + key.getGroup(); + String triggerType = SchedulerServiceConstants.TRIGGER_TYPE_CRON; + if (context.getMergedJobDataMap().containsKey(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE)) { + triggerType = context.getMergedJobDataMap().getString(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE); + } + return schedularService.processJobDetailForExecution(jobKey, triggerType); + } + + @Override + public void triggerMisfired(@SuppressWarnings("unused") Trigger trigger) { + + } + + @Override + public void triggerComplete(@SuppressWarnings("unused") Trigger trigger, @SuppressWarnings("unused") JobExecutionContext context, + @SuppressWarnings("unused") CompletedExecutionInstruction triggerInstructionCode) { + + } + +}