Merge pull request #475 from pramodn02/MIFOSX-506_77

added scheduler groups for dependent jobs

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`;

mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/domain/ScheduledJobDetail.java

@@ -62,6 +62,13 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
     @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<Long> {
         return this.jobKey;
     }
 
+    public Short getSchedulerGroup() {
+        return this.schedulerGroup;
+    }
+
     public boolean isActiveSchedular() {
         return this.activeSchedular;
     }

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();
+
 }

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<Long, Boolean> schedulerStatus = new HashMap<Long, Boolean>(2);
+
+    private final HashMap<String, Scheduler> schedulers = new HashMap<String, Scheduler>(4);
+
     @Autowired
-    public JobRegisterServiceImpl(final ApplicationContext applicationContext, final SchedulerFactoryBean schedulerFactoryBean,
+    public JobRegisterServiceImpl(final ApplicationContext applicationContext, final SchedularWritePlatformService schedularService,
-            final SchedularWritePlatformService schedularService, final TenantDetailsService tenantDetailsService) {
+            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<ScheduledJobDetail> 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);
+            String schedulername = getSchedulerName(scheduledJobDetail);
-            scheduleJob(scheduledJobDetails);
+            Scheduler scheduler = schedulers.get(schedulername);
-            this.schedularWritePlatformService.saveOrUpdate(scheduledJobDetails);
+            if (scheduler != null) {
+                scheduler.deleteJob(jobKey);
+            }
+            scheduleJob(scheduledJobDetail);
+            this.schedularWritePlatformService.saveOrUpdate(scheduledJobDetail);
         } catch (Throwable throwable) {
             String stackTrace = getStackTraceAsString(throwable);
-            scheduledJobDetails.updateErrorLog(stackTrace);
+            scheduledJobDetail.updateErrorLog(stackTrace);
-            this.schedularWritePlatformService.saveOrUpdate(scheduledJobDetails);
+            this.schedularWritePlatformService.saveOrUpdate(scheduledJobDetail);
+        }
+    }
+
+    @Override
+    public void stopScheduler() {
+        schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), false);
+        for (Map.Entry<String, Scheduler> 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<String, Scheduler> 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();
+                executeJob(scheduledJobDetail);
-                JobKey jobKey = constructJobKey(key);
-                executeJob(jobKey);
             } 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 = getBeanObject(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;
-            }
-        }
         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

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;
+    
 }

mifosng-provider/src/main/resources/META-INF/spring/appContext.xml

@@ -51,5 +51,4 @@
 		
 	<import resource="securityContext.xml" />
 
-    <import resource="scheduler.xml"/>
 </beans>

mifosng-provider/src/main/resources/META-INF/spring/scheduler.xml

@@ -1,22 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
-
-    This Source Code Form is subject to the terms of the Mozilla Public
-    License, v. 2.0. If a copy of the MPL was not distributed with this file,
-    You can obtain one at http://mozilla.org/MPL/2.0/.
-
--->
-<beans xmlns="http://www.springframework.org/schema/beans"
-	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
-
-	<bean name="schedulerFactory"
-		class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
-		<property name="globalJobListeners">
-			<list>
-				<ref bean = "schedulerJobListener"/>
-			</list>
-		</property>
-	</bean>
-	
-</beans>