diff --git a/mifosng-db/migrations/core_db/V89__added_scheduler_group.sql b/mifosng-db/migrations/core_db/V89__added_scheduler_group.sql new file mode 100755 index 000000000..95f6f1c2a --- /dev/null +++ b/mifosng-db/migrations/core_db/V89__added_scheduler_group.sql @@ -0,0 +1,2 @@ +ALTER TABLE `job` + ADD COLUMN `scheduler_group` SMALLINT(2) NOT NULL DEFAULT '0' AFTER `updates_allowed`; \ No newline at end of file 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 8d3cecec8..ac391a344 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 @@ -62,6 +62,13 @@ 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; + protected ScheduledJobDetail() { } @@ -86,6 +93,10 @@ public class ScheduledJobDetail extends AbstractPersistable { return this.jobKey; } + public Short getSchedulerGroup() { + return this.schedulerGroup; + } + public boolean isActiveSchedular() { return this.activeSchedular; } diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterService.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterService.java index d89f82d41..44d0cf426 100755 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterService.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterService.java @@ -6,4 +6,10 @@ public interface JobRegisterService { public void rescheduleJob(Long jobId); + public void stopScheduler(); + + public void startScheduler(); + + public boolean isSchedulerRunning(); + } 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 06190c20f..0cf1b06a3 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 @@ -1,6 +1,9 @@ package org.mifosplatform.infrastructure.jobs.service; +import java.util.HashMap; import java.util.List; +import java.util.Map; +import java.util.Properties; import java.util.TimeZone; import javax.annotation.PostConstruct; @@ -14,6 +17,8 @@ import org.mifosplatform.infrastructure.security.service.TenantDetailsService; import org.quartz.JobDataMap; import org.quartz.JobDetail; import org.quartz.JobKey; +import org.quartz.JobListener; +import org.quartz.Scheduler; import org.quartz.SchedulerException; import org.quartz.Trigger; import org.slf4j.Logger; @@ -38,18 +43,19 @@ public class JobRegisterServiceImpl implements JobRegisterService { private final ApplicationContext applicationContext; - private final SchedulerFactoryBean schedulerFactoryBean; - private final SchedularWritePlatformService schedularWritePlatformService; private final TenantDetailsService tenantDetailsService; + private final HashMap schedulerStatus = new HashMap(2); + + private final HashMap schedulers = new HashMap(4); + @Autowired - public JobRegisterServiceImpl(final ApplicationContext applicationContext, final SchedulerFactoryBean schedulerFactoryBean, - final SchedularWritePlatformService schedularService, final TenantDetailsService tenantDetailsService) { + public JobRegisterServiceImpl(final ApplicationContext applicationContext, final SchedularWritePlatformService schedularService, + final TenantDetailsService tenantDetailsService) { this.applicationContext = applicationContext; this.schedularWritePlatformService = schedularService; - this.schedulerFactoryBean = schedulerFactoryBean; this.tenantDetailsService = tenantDetailsService; } @@ -59,6 +65,7 @@ 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); this.schedularWritePlatformService.saveOrUpdate(jobDetails); @@ -66,14 +73,18 @@ public class JobRegisterServiceImpl implements JobRegisterService { } } - public void executeJob(final JobKey jobKey) { + public void executeJob(final ScheduledJobDetail scheduledJobDetail) { try { final JobDataMap jobDataMap = new JobDataMap(); jobDataMap.put(SchedularServiceConstants.TRIGGER_TYPE_REFERENCE, SchedularServiceConstants.TRIGGER_TYPE_APPLICATION); jobDataMap.put(SchedularServiceConstants.TENANT_IDENTIFIER, ThreadLocalContextUtil.getTenant().getTenantIdentifier()); - if (schedulerFactoryBean.getScheduler().checkExists(jobKey)) { + String key = scheduledJobDetail.getJobKey(); + JobKey jobKey = constructJobKey(key); + String schedulerName = getSchedulerName(scheduledJobDetail); + Scheduler scheduler = schedulers.get(schedulerName); + if (scheduler != null && scheduler.checkExists(jobKey)) { try { - schedulerFactoryBean.getScheduler().triggerJob(jobKey, jobDataMap); + scheduler.triggerJob(jobKey, jobDataMap); } catch (SchedulerException e) { logger.error(e.getMessage(), e); } @@ -86,17 +97,51 @@ public class JobRegisterServiceImpl implements JobRegisterService { } - public void rescheduleJob(ScheduledJobDetail scheduledJobDetails) { + public void rescheduleJob(ScheduledJobDetail scheduledJobDetail) { try { - String jobIdentity = scheduledJobDetails.getJobKey(); + String jobIdentity = scheduledJobDetail.getJobKey(); JobKey jobKey = constructJobKey(jobIdentity); - schedulerFactoryBean.getScheduler().deleteJob(jobKey); - scheduleJob(scheduledJobDetails); - this.schedularWritePlatformService.saveOrUpdate(scheduledJobDetails); + String schedulername = getSchedulerName(scheduledJobDetail); + Scheduler scheduler = schedulers.get(schedulername); + if (scheduler != null) { + scheduler.deleteJob(jobKey); + } + scheduleJob(scheduledJobDetail); + this.schedularWritePlatformService.saveOrUpdate(scheduledJobDetail); } catch (Throwable throwable) { String stackTrace = getStackTraceAsString(throwable); - scheduledJobDetails.updateErrorLog(stackTrace); - this.schedularWritePlatformService.saveOrUpdate(scheduledJobDetails); + scheduledJobDetail.updateErrorLog(stackTrace); + this.schedularWritePlatformService.saveOrUpdate(scheduledJobDetail); + } + } + + @Override + public void stopScheduler() { + 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); + } + } + } + + @Override + public void startScheduler() { + schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), true); + for (Map.Entry schedulerEntry : this.schedulers.entrySet()) { + try { + String schedulerName = schedulerEntry.getValue().getSchedulerName(); + if (isCurrentTenantsScheduler(schedulerName)) { + schedulerEntry.getValue().start(); + } + } catch (SchedulerException e) { + logger.error(e.getMessage(), e); + } } } @@ -111,9 +156,7 @@ public class JobRegisterServiceImpl implements JobRegisterService { try { ScheduledJobDetail scheduledJobDetail = this.schedularWritePlatformService.findByJobId(jobId); if (scheduledJobDetail != null) { - String key = scheduledJobDetail.getJobKey(); - JobKey jobKey = constructJobKey(key); - executeJob(jobKey); + executeJob(scheduledJobDetail); } else { throw new JobNotFoundException(String.valueOf(jobId)); } @@ -122,6 +165,22 @@ public class JobRegisterServiceImpl implements JobRegisterService { } } + @Override + public boolean isSchedulerRunning() { + return this.schedulerStatus.get(ThreadLocalContextUtil.getTenant().getId()); + } + + private boolean isCurrentTenantsScheduler(String schedulerName) { + int beginIndex = SchedularServiceConstants.SCHEDULER.length(); + int endIndex = schedulerName.length(); + if (schedulerName.contains(SchedularServiceConstants.SCHEDULER_GROUP)) { + endIndex = schedulerName.indexOf(SchedularServiceConstants.SCHEDULER_GROUP); + } + String tenantId = schedulerName.substring(beginIndex, endIndex); + MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant(); + return tenant.getId() == Long.valueOf(tenantId); + } + private void scheduleJob(ScheduledJobDetail scheduledJobDetails) { if (!scheduledJobDetails.isActiveSchedular()) { scheduledJobDetails.updateNextRunTime(null); @@ -129,9 +188,10 @@ public class JobRegisterServiceImpl implements JobRegisterService { return; } try { + Scheduler scheduler = getScheduler(scheduledJobDetails); JobDetail jobDetail = createJobDetail(scheduledJobDetails); Trigger trigger = createTrigger(scheduledJobDetails, jobDetail); - schedulerFactoryBean.getScheduler().scheduleJob(jobDetail, trigger); + scheduler.scheduleJob(jobDetail, trigger); scheduledJobDetails.updateJobKey(getJobKeyAsString(jobDetail.getKey())); scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime()); scheduledJobDetails.updateErrorLog(null); @@ -143,17 +203,52 @@ public class JobRegisterServiceImpl implements JobRegisterService { scheduledJobDetails.updateCurrentlyRunningStatus(false); } + private Scheduler getScheduler(ScheduledJobDetail scheduledJobDetail) throws Exception { + String schedulername = getSchedulerName(scheduledJobDetail); + Scheduler scheduler = this.schedulers.get(schedulername); + if (scheduler == null) { + int noOfThreads = SchedularServiceConstants.DEFAULT_THREAD_COUNT; + if (scheduledJobDetail.getSchedulerGroup() > 0) { + noOfThreads = SchedularServiceConstants.GROUP_THREAD_COUNT; + } + scheduler = createScheduler(schedulername, noOfThreads); + this.schedulers.put(schedulername, scheduler); + } + return scheduler; + } + + private String getSchedulerName(ScheduledJobDetail scheduledJobDetail) { + StringBuilder sb = new StringBuilder(20); + MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant(); + sb.append(SchedularServiceConstants.SCHEDULER).append(tenant.getId()); + if (scheduledJobDetail.getSchedulerGroup() > 0) { + sb.append(SchedularServiceConstants.SCHEDULER_GROUP).append(scheduledJobDetail.getSchedulerGroup()); + } + return sb.toString(); + } + + private Scheduler createScheduler(String name, int noOfThreads) throws Exception { + SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean(); + schedulerFactoryBean.setSchedulerName(name); + schedulerFactoryBean.setGlobalJobListeners(getGlobalListener()); + Properties quartzProperties = new Properties(); + quartzProperties.put(SchedulerFactoryBean.PROP_THREAD_COUNT, Integer.toString(noOfThreads)); + schedulerFactoryBean.setQuartzProperties(quartzProperties); + schedulerFactoryBean.afterPropertiesSet(); + schedulerFactoryBean.start(); + return schedulerFactoryBean.getScheduler(); + } + + private JobListener[] getGlobalListener() throws ClassNotFoundException { + JobListener listener = (JobListener) getBeanObject(SchedulerJobListener.class); + JobListener[] listenerArray = { listener }; + return listenerArray; + } + private JobDetail createJobDetail(ScheduledJobDetail scheduledJobDetails) throws Exception { MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant(); String[] jobDetails = CronMethodParser.findTargetMethodDetails(scheduledJobDetails.getJobName()); - String[] beanNames = this.applicationContext.getBeanNamesForType(Class.forName(jobDetails[CronMethodParser.CLASS_INDEX])); - Object targetObject = this.applicationContext.getBean(beanNames[0]); - for (String beanName : beanNames) { - Object nextObject = this.applicationContext.getBean(beanName); - if (nextObject != null && !targetObject.getClass().isAssignableFrom(nextObject.getClass())) { - targetObject = nextObject; - } - } + Object targetObject = getBeanObject(Class.forName(jobDetails[CronMethodParser.CLASS_INDEX])); MethodInvokingJobDetailFactoryBean jobDetailFactoryBean = new MethodInvokingJobDetailFactoryBean(); jobDetailFactoryBean.setName(scheduledJobDetails.getJobName() + "JobDetail" + tenant.getId()); jobDetailFactoryBean.setTargetObject(targetObject); @@ -164,6 +259,23 @@ public class JobRegisterServiceImpl implements JobRegisterService { return jobDetailFactoryBean.getObject(); } + /** + * @param jobDetails + * @return + * @throws ClassNotFoundException + */ + private Object getBeanObject(Class classType) throws ClassNotFoundException { + String[] beanNames = this.applicationContext.getBeanNamesForType(classType); + Object targetObject = this.applicationContext.getBean(beanNames[0]); + for (String beanName : beanNames) { + Object nextObject = this.applicationContext.getBean(beanName); + if (nextObject != null && !targetObject.getClass().isAssignableFrom(nextObject.getClass())) { + targetObject = nextObject; + } + } + return targetObject; + } + /** * @param scheduledJobDetails * @return diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularServiceConstants.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularServiceConstants.java index 35c44bfb8..46866c99f 100644 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularServiceConstants.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularServiceConstants.java @@ -14,4 +14,9 @@ public interface SchedularServiceConstants { public static final String DEFAULT_LISTENER_NAME = "Global Listner"; public static final int STACK_TRACE_LEVEL = 7; public static final String TENANT_IDENTIFIER = "tenantIdentifier"; + public static final String SCHEDULER = "Scheduler"; + public static final String SCHEDULER_GROUP = "group"; + public static final int DEFAULT_THREAD_COUNT = 7; + public static final int GROUP_THREAD_COUNT = 1; + } \ No newline at end of file diff --git a/mifosng-provider/src/main/resources/META-INF/spring/appContext.xml b/mifosng-provider/src/main/resources/META-INF/spring/appContext.xml index e6dde35ea..6ebf89449 100644 --- a/mifosng-provider/src/main/resources/META-INF/spring/appContext.xml +++ b/mifosng-provider/src/main/resources/META-INF/spring/appContext.xml @@ -51,5 +51,4 @@ - \ No newline at end of file diff --git a/mifosng-provider/src/main/resources/META-INF/spring/scheduler.xml b/mifosng-provider/src/main/resources/META-INF/spring/scheduler.xml deleted file mode 100644 index 4302beb10..000000000 --- a/mifosng-provider/src/main/resources/META-INF/spring/scheduler.xml +++ /dev/null @@ -1,22 +0,0 @@ - - - - - - - - - - - - - \ No newline at end of file