Merge pull request #455 from pramodn02/MIFOSX-506_7

scheduler changes for API
This commit is contained in:
Vishwas Babu A J
2013-07-16 03:31:11 -07:00
18 changed files with 640 additions and 194 deletions
@@ -0,0 +1,5 @@
ALTER TABLE `job`
ADD COLUMN `currently_running` TINYINT(1) NOT NULL DEFAULT '0' AFTER `is_active`;
ALTER TABLE `job`
CHANGE COLUMN `trigger_key` `job_key` VARCHAR(500) NULL DEFAULT NULL;
@@ -0,0 +1,34 @@
package org.mifosplatform.infrastructure.jobs.api;
import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
public class SchedulerJobApiConstants {
// response parameters
public static final String jobIdentifierParamName = "jobId";
public static final String displayNameParamName = "displayName";
public static final String nextRunTimeParamName = "nextRunTime";
public static final String initializingErrorParamName = "initializingError";
public static final String jobActiveStatusParamName = "active";
public static final String currentlyRunningParamName = "currentlyRunning";
public static final String lastRunHistoryObjParamName = "lastRunHistory";
public static final String versionParamName = "version";
public static final String jobRunStartTimeParamName = "jobRunStartTime";
public static final String jobRunEndTimeParamName = "jobRunEndTime";
public static final String statusParamName = "status";
public static final String jobRunErrorMessageParamName = "jobRunErrorMessage";
public static final String triggerTypeParamName = "triggerType";
public static final String jobRunErrorLogParamName = "jobRunErrorLog";
public static final Set<String> JOB_DETAIL_RESPONSE_DATA_PARAMETERS = new HashSet<String>(Arrays.asList(jobIdentifierParamName,
displayNameParamName, nextRunTimeParamName, initializingErrorParamName, jobActiveStatusParamName,currentlyRunningParamName, lastRunHistoryObjParamName));
public static final Set<String> JOB_HISTORY_RESPONSE_DATA_PARAMETERS = new HashSet<String>(Arrays.asList(versionParamName,
jobRunStartTimeParamName, jobRunEndTimeParamName, statusParamName, jobRunErrorMessageParamName, triggerTypeParamName,
jobRunErrorLogParamName));
}
@@ -0,0 +1,89 @@
package org.mifosplatform.infrastructure.jobs.api;
import java.util.List;
import javax.ws.rs.Consumes;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
import javax.ws.rs.Produces;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriInfo;
import org.mifosplatform.infrastructure.core.api.ApiRequestParameterHelper;
import org.mifosplatform.infrastructure.core.serialization.ApiRequestJsonSerializationSettings;
import org.mifosplatform.infrastructure.core.serialization.ToApiJsonSerializer;
import org.mifosplatform.infrastructure.core.service.Page;
import org.mifosplatform.infrastructure.jobs.data.JobDetailData;
import org.mifosplatform.infrastructure.jobs.data.JobDetailHistoryData;
import org.mifosplatform.infrastructure.jobs.service.JobRegisterService;
import org.mifosplatform.infrastructure.jobs.service.SchedulerJobRunnerReadService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Path("/schedulerJobs")
@Component
public class SchedulerJobApiResource {
private final SchedulerJobRunnerReadService schedulerJobRunnerReadService;
private final JobRegisterService jobRegisterService;
private final ApiRequestParameterHelper apiRequestParameterHelper;
private final ToApiJsonSerializer<JobDetailData> toApiJsonSerializer;
private final ToApiJsonSerializer<JobDetailHistoryData> jobHistoryToApiJsonSerializer;
@Autowired
public SchedulerJobApiResource(final SchedulerJobRunnerReadService schedulerJobRunnerReadService,
final JobRegisterService jobRegisterService, final ToApiJsonSerializer<JobDetailData> toApiJsonSerializer,
final ApiRequestParameterHelper apiRequestParameterHelper,
final ToApiJsonSerializer<JobDetailHistoryData> jobHistoryToApiJsonSerializer) {
this.schedulerJobRunnerReadService = schedulerJobRunnerReadService;
this.jobRegisterService = jobRegisterService;
this.toApiJsonSerializer = toApiJsonSerializer;
this.jobHistoryToApiJsonSerializer = jobHistoryToApiJsonSerializer;
this.apiRequestParameterHelper = apiRequestParameterHelper;
}
@GET
@Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON })
public String retrieveAll(@Context final UriInfo uriInfo) {
List<JobDetailData> jobDetailDatas = this.schedulerJobRunnerReadService.findAllJobDeatils();
final ApiRequestJsonSerializationSettings settings = this.apiRequestParameterHelper.process(uriInfo.getQueryParameters());
return this.toApiJsonSerializer.serialize(settings, jobDetailDatas, SchedulerJobApiConstants.JOB_DETAIL_RESPONSE_DATA_PARAMETERS);
}
@GET
@Path("{jobId}")
@Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON })
public String retrieveOne(@PathParam("jobId") final Long jobId, @Context final UriInfo uriInfo) {
JobDetailData jobDetailData = this.schedulerJobRunnerReadService.retrieveOne(jobId);
final ApiRequestJsonSerializationSettings settings = apiRequestParameterHelper.process(uriInfo.getQueryParameters());
return this.toApiJsonSerializer.serialize(settings, jobDetailData, SchedulerJobApiConstants.JOB_DETAIL_RESPONSE_DATA_PARAMETERS);
}
@GET
@Path("{jobId}/history")
@Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON })
public String retrieveHistory(@Context final UriInfo uriInfo, @PathParam("jobId") final Long jobId) {
Page<JobDetailHistoryData> jobhistoryDetailData = this.schedulerJobRunnerReadService.retrieveJobHistory(jobId);
final ApiRequestJsonSerializationSettings settings = apiRequestParameterHelper.process(uriInfo.getQueryParameters());
return this.jobHistoryToApiJsonSerializer.serialize(settings, jobhistoryDetailData,
SchedulerJobApiConstants.JOB_HISTORY_RESPONSE_DATA_PARAMETERS);
}
@POST
@Path("{jobId}/executeJob")
@Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON })
public Response executeJob(@PathParam("jobId") final Long jobId) {
Response response = Response.status(400).build();
jobRegisterService.executeJob(jobId);
response = Response.status(202).build();
return response;
}
}
@@ -0,0 +1,32 @@
package org.mifosplatform.infrastructure.jobs.data;
import java.util.Date;
public class JobDetailData {
private Long jobId;
private String displayName;
private Date nextRunTime;
private String initializingError;
private boolean active;
private boolean currentlyRunning;
private JobDetailHistoryData lastRunHistory;
public JobDetailData(final Long jobId, final String displayName, final Date nextRunTime, final String initializingError,
final boolean active, final boolean currentlyRunning, final JobDetailHistoryData lastRunHistory) {
this.jobId = jobId;
this.displayName = displayName;
this.nextRunTime = nextRunTime;
this.initializingError = initializingError;
this.active = active;
this.lastRunHistory = lastRunHistory;
this.currentlyRunning = currentlyRunning;
}
}
@@ -0,0 +1,25 @@
package org.mifosplatform.infrastructure.jobs.data;
import java.util.Date;
public class JobDetailHistoryData {
private Long version;
private Date jobRunStartTime;
private Date jobRunEndTime;
private String status;
private String jobRunErrorMessage;
private String triggerType;
private String jobRunErrorLog;
public JobDetailHistoryData(final Long version, final Date jobRunStartTime, final Date jobRunEndTime, final String status,
final String jobRunErrorMessage, final String triggerType, final String jobRunErrorLog) {
this.version = version;
this.jobRunStartTime = jobRunStartTime;
this.jobRunEndTime = jobRunEndTime;
this.status = status;
this.jobRunErrorMessage = jobRunErrorMessage;
this.triggerType = triggerType;
this.jobRunErrorLog = jobRunErrorLog;
}
}
@@ -45,8 +45,8 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
@Temporal(TemporalType.TIMESTAMP)
private Date nextRunTime;
@Column(name = "trigger_key")
private String triggerKey;
@Column(name = "job_key")
private String jobKey;
@SuppressWarnings("unused")
@Column(name = "initializing_errorlog")
@@ -55,13 +55,16 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
@Column(name = "is_active")
private boolean activeSchedular;
@Column(name = "currently_running")
private boolean currentlyRunning;
public ScheduledJobDetail() {
}
public ScheduledJobDetail(final String jobName, final String jobDisplayName, final String croneExpression, final Date createTime,
final Short taskPriority, final String groupName, final Date previousRunStartTime, final Date nextRunTime,
final String triggKey, final String errorLog, final boolean activeSchedular) {
public ScheduledJobDetail(final String jobName, final String jobDisplayName, final String croneExpression, Date createTime,
final Short taskPriority, final String groupName, final Date previousRunStartTime, final Date nextRunTime, final String jobKey,
final String errorLog, final boolean activeSchedular, final boolean currentlyRunning) {
this.jobName = jobName;
this.jobDisplayName = jobDisplayName;
this.croneExpression = croneExpression;
@@ -70,9 +73,10 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
this.groupName = groupName;
this.previousRunStartTime = previousRunStartTime;
this.nextRunTime = nextRunTime;
this.triggerKey = triggKey;
this.jobKey = jobKey;
this.errorLog = errorLog;
this.activeSchedular = activeSchedular;
this.currentlyRunning = currentlyRunning;
}
public String getJobName() {
@@ -91,8 +95,8 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
return this.groupName;
}
public String getTriggerKey() {
return this.triggerKey;
public String getJobKey() {
return this.jobKey;
}
public boolean isActiveSchedular() {
@@ -111,12 +115,20 @@ public class ScheduledJobDetail extends AbstractPersistable<Long> {
this.nextRunTime = nextRunTime;
}
public void updateTriggerKey(final String triggerKey) {
this.triggerKey = triggerKey;
public void updateJobKey(String jobKey) {
this.jobKey = jobKey;
}
public void updateErrorLog(final String errorLog) {
this.errorLog = errorLog;
}
public boolean isCurrentlyRunning() {
return this.currentlyRunning;
}
public void updateCurrentlyRunningStatus(boolean currentlyRunning) {
this.currentlyRunning = currentlyRunning;
}
}
@@ -8,7 +8,10 @@ import org.springframework.data.repository.query.Param;
public interface ScheduledJobDetailRepository extends JpaRepository<ScheduledJobDetail, Long>,
JpaSpecificationExecutor<ScheduledJobDetail> {
@Query("from ScheduledJobDetail jobDetail where jobDetail.triggerKey = :triggerKey")
ScheduledJobDetail findByTriggerKey(@Param("triggerKey") String triggerKey);
@Query("from ScheduledJobDetail jobDetail where jobDetail.jobKey = :jobKey")
ScheduledJobDetail findByJobKey(@Param("jobKey") String jobKey);
@Query("from ScheduledJobDetail jobDetail where jobDetail.id=:jobId")
ScheduledJobDetail findByJobId(@Param("jobId") Long jobId);
}
@@ -8,7 +8,7 @@ import org.springframework.data.repository.query.Param;
public interface ScheduledJobRunHistoryRepository extends JpaRepository<ScheduledJobRunHistory, Long>,
JpaSpecificationExecutor<ScheduledJobRunHistory> {
@Query("select max(sjrh.version) from ScheduledJobRunHistory sjrh where sjrh.scheduledJobDetail.triggerKey = :triggerKey")
Long findMaxVersionByTriggerKey(@Param("triggerKey") String triggerKey);
@Query("select max(sjrh.version) from ScheduledJobRunHistory sjrh where sjrh.scheduledJobDetail.jobKey = :jobKey")
Long findMaxVersionByJobKey(@Param("jobKey") String jobKey);
}
@@ -0,0 +1,12 @@
package org.mifosplatform.infrastructure.jobs.exception;
import org.mifosplatform.infrastructure.core.exception.AbstractPlatformResourceNotFoundException;
public class JobInProcessExecution extends AbstractPlatformResourceNotFoundException {
public JobInProcessExecution(String identifier) {
super("error.msg.sheduler.job.inprogress", "job execution is in process for "+identifier, identifier);
}
}
@@ -0,0 +1,18 @@
/**
* 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/.
*/
package org.mifosplatform.infrastructure.jobs.exception;
import org.mifosplatform.infrastructure.core.exception.AbstractPlatformResourceNotFoundException;
/**
* A {@link RuntimeException} thrown when Job resources are not found.
*/
public class JobNotFoundException extends AbstractPlatformResourceNotFoundException {
public JobNotFoundException(final String identifier) {
super("error.msg.sheduler.job.id.invalid", "Job with identifier " + identifier + " does not exist", identifier);
}
}
@@ -1,172 +1,11 @@
package org.mifosplatform.infrastructure.jobs.service;
import java.util.List;
import java.util.TimeZone;
import javax.annotation.PostConstruct;
import org.mifosplatform.infrastructure.core.domain.MifosPlatformTenant;
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.security.service.TenantDetailsService;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.TriggerKey;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.quartz.CronTriggerFactoryBean;
import org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;
import org.springframework.stereotype.Service;
/**
* Service class to create and load batch jobs to Scheduler using
* {@link SchedulerFactoryBean} ,{@link MethodInvokingJobDetailFactoryBean} and
* {@link CronTriggerFactoryBean}
*
*/
@Service
public class JobRegisterService {
public interface JobRegisterService {
private final ApplicationContext applicationContext;
public void executeJob(Long jobId);
private final SchedulerFactoryBean schedulerFactoryBean;
private final SchedularService schedularService;
private final TenantDetailsService tenantDetailsService;
@Autowired
public JobRegisterService(final ApplicationContext applicationContext, final SchedulerFactoryBean schedulerFactoryBean,
final SchedularService schedularService, final TenantDetailsService tenantDetailsService) {
this.applicationContext = applicationContext;
this.schedularService = schedularService;
this.schedulerFactoryBean = schedulerFactoryBean;
this.tenantDetailsService = tenantDetailsService;
}
@PostConstruct
public void loadAllJobs() {
List<MifosPlatformTenant> allTenants = this.tenantDetailsService.findAllTenants();
for (MifosPlatformTenant tenant : allTenants) {
ThreadLocalContextUtil.setTenant(tenant);
List<ScheduledJobDetail> scheduledJobDetails = this.schedularService.getScheduledJobDetails();
for (ScheduledJobDetail jobDetails : scheduledJobDetails) {
if (jobDetails.isActiveSchedular()) {
scheduleJob(jobDetails);
this.schedularService.saveOrUpdate(jobDetails);
}
}
}
}
public void excuteJob(TriggerKey triggerKey) {
JobDataMap jobDataMap = new JobDataMap();
jobDataMap.put(SchedularServiceConstants.TRIGGER_TYPE_REFERENCE, SchedularServiceConstants.TRIGGER_TYPE_APPLICATION);
try {
JobKey jobKey = schedulerFactoryBean.getScheduler().getTrigger(triggerKey).getJobKey();
schedulerFactoryBean.getScheduler().triggerJob(jobKey);
} catch (SchedulerException e) {
e.printStackTrace();
}
}
public void rescheduleJob(ScheduledJobDetail scheduledJobDetails) {
try {
String triggerIdentity = scheduledJobDetails.getTriggerKey();
String[] triggerKeyIds = triggerIdentity.split(SchedularServiceConstants.TRIGGER_KEY_SEPERATOR);
TriggerKey triggerKey = new TriggerKey(triggerKeyIds[0].trim(), triggerKeyIds[1].trim());
JobDetail jobDetail = createJobDetail(scheduledJobDetails);
Trigger trigger = createTrigger(scheduledJobDetails, jobDetail);
schedulerFactoryBean.getScheduler().rescheduleJob(triggerKey, trigger);
scheduledJobDetails.updateTriggerKey(getTriggerKey(trigger.getKey()));
scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime());
scheduledJobDetails.updateErrorLog(null);
this.schedularService.saveOrUpdate(scheduledJobDetails);
} catch (Throwable throwable) {
String stackTrace = getStackTraceAsString(throwable);
scheduledJobDetails.updateErrorLog(stackTrace);
this.schedularService.saveOrUpdate(scheduledJobDetails);
}
}
private void scheduleJob(ScheduledJobDetail scheduledJobDetails) {
try {
JobDetail jobDetail = createJobDetail(scheduledJobDetails);
Trigger trigger = createTrigger(scheduledJobDetails, jobDetail);
schedulerFactoryBean.getScheduler().scheduleJob(jobDetail, trigger);
scheduledJobDetails.updateTriggerKey(getTriggerKey(trigger.getKey()));
scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime());
scheduledJobDetails.updateErrorLog(null);
} catch (Throwable throwable) {
String stackTrace = getStackTraceAsString(throwable);
scheduledJobDetails.updateErrorLog(stackTrace);
}
}
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;
}
}
MethodInvokingJobDetailFactoryBean jobDetailFactoryBean = new MethodInvokingJobDetailFactoryBean();
jobDetailFactoryBean.setName(scheduledJobDetails.getJobName() + "JobDetail-" + tenant.getId());
jobDetailFactoryBean.setTargetObject(targetObject);
jobDetailFactoryBean.setTargetMethod(jobDetails[CronMethodParser.METHOD_INDEX]);
jobDetailFactoryBean.setGroup(scheduledJobDetails.getGroupName());
jobDetailFactoryBean.setConcurrent(false);
jobDetailFactoryBean.afterPropertiesSet();
return jobDetailFactoryBean.getObject();
}
/**
* @param scheduledJobDetails
* @return
*/
private Trigger createTrigger(ScheduledJobDetail scheduledJobDetails, JobDetail jobDetail) {
MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant();
CronTriggerFactoryBean cronTriggerFactoryBean = new CronTriggerFactoryBean();
cronTriggerFactoryBean.setName(scheduledJobDetails.getJobName() + "Trigger-" + tenant.getId());
cronTriggerFactoryBean.setJobDetail(jobDetail);
JobDataMap jobDataMap = new JobDataMap();
jobDataMap.put(SchedularServiceConstants.TENANT_IDENTIFIER, tenant.getTenantIdentifier());
cronTriggerFactoryBean.setJobDataMap(jobDataMap);
TimeZone timeZone = TimeZone.getTimeZone(tenant.getTimezoneId());
cronTriggerFactoryBean.setTimeZone(timeZone);
cronTriggerFactoryBean.setGroup(scheduledJobDetails.getGroupName());
cronTriggerFactoryBean.setCronExpression(scheduledJobDetails.getCroneExpression());
cronTriggerFactoryBean.setPriority(scheduledJobDetails.getTaskPriority());
cronTriggerFactoryBean.afterPropertiesSet();
return cronTriggerFactoryBean.getObject();
}
/**
* @param throwable
* @return
*/
private String getStackTraceAsString(Throwable throwable) {
StackTraceElement[] stackTraceElements = throwable.getStackTrace();
StringBuffer sb = new StringBuffer(throwable.toString());
for (StackTraceElement element : stackTraceElements) {
sb.append("\n \t at ").append(element.getClassName()).append(".").append(element.getMethodName()).append("(")
.append(element.getLineNumber()).append(")");
}
return sb.toString();
}
private String getTriggerKey(TriggerKey triggerKey) {
return triggerKey.getName() + SchedularServiceConstants.TRIGGER_KEY_SEPERATOR + triggerKey.getGroup();
}
public void rescheduleJob(ScheduledJobDetail scheduledJobDetails);
}
@@ -0,0 +1,217 @@
package org.mifosplatform.infrastructure.jobs.service;
import java.util.List;
import java.util.TimeZone;
import javax.annotation.PostConstruct;
import org.mifosplatform.infrastructure.core.domain.MifosPlatformTenant;
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.exception.JobNotFoundException;
import org.mifosplatform.infrastructure.security.service.TenantDetailsService;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.quartz.CronTriggerFactoryBean;
import org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;
import org.springframework.stereotype.Service;
/**
* Service class to create and load batch jobs to Scheduler using
* {@link SchedulerFactoryBean} ,{@link MethodInvokingJobDetailFactoryBean} and
* {@link CronTriggerFactoryBean}
*
*/
@Service
public class JobRegisterServiceImpl implements JobRegisterService {
private final static Logger logger = LoggerFactory.getLogger(JobRegisterServiceImpl.class);
private final ApplicationContext applicationContext;
private final SchedulerFactoryBean schedulerFactoryBean;
private final SchedularService schedularService;
private final TenantDetailsService tenantDetailsService;
@Autowired
public JobRegisterServiceImpl(final ApplicationContext applicationContext, final SchedulerFactoryBean schedulerFactoryBean,
final SchedularService schedularService, final TenantDetailsService tenantDetailsService) {
this.applicationContext = applicationContext;
this.schedularService = schedularService;
this.schedulerFactoryBean = schedulerFactoryBean;
this.tenantDetailsService = tenantDetailsService;
}
@PostConstruct
public void loadAllJobs() {
List<MifosPlatformTenant> allTenants = this.tenantDetailsService.findAllTenants();
for (MifosPlatformTenant tenant : allTenants) {
ThreadLocalContextUtil.setTenant(tenant);
List<ScheduledJobDetail> scheduledJobDetails = this.schedularService.retrieveAllJobs();
for (ScheduledJobDetail jobDetails : scheduledJobDetails) {
if (jobDetails.isActiveSchedular()) {
scheduleJob(jobDetails);
}else{
jobDetails.updateNextRunTime(null);
}
this.schedularService.saveOrUpdate(jobDetails);
}
}
}
public void executeJob(final JobKey jobKey) {
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)) {
Thread newThread = new Thread(new Runnable() {
@Override
public void run() {
final JobKey threadLocalJobKey = jobKey;
final JobDataMap threadLocalJobDataMap = jobDataMap;
try {
schedulerFactoryBean.getScheduler().triggerJob(threadLocalJobKey, threadLocalJobDataMap);
} catch (SchedulerException e) {
logger.error(e.getMessage(), e);
}
}
});
newThread.run();
} else {
throw new JobNotFoundException(jobKey.toString());
}
} catch (SchedulerException e) {
logger.error(e.getMessage(), e);
}
}
@Override
public void rescheduleJob(ScheduledJobDetail scheduledJobDetails) {
try {
String jobIdentity = scheduledJobDetails.getJobKey();
JobKey jobKey = constructJobKey(jobIdentity);
schedulerFactoryBean.getScheduler().deleteJob(jobKey);
scheduleJob(scheduledJobDetails);
this.schedularService.saveOrUpdate(scheduledJobDetails);
} catch (Throwable throwable) {
String stackTrace = getStackTraceAsString(throwable);
scheduledJobDetails.updateErrorLog(stackTrace);
this.schedularService.saveOrUpdate(scheduledJobDetails);
}
}
@Override
public void executeJob(Long jobId) {
try {
ScheduledJobDetail scheduledJobDetail = this.schedularService.findByJobId(jobId);
if (scheduledJobDetail != null) {
String key = scheduledJobDetail.getJobKey();
JobKey jobKey = constructJobKey(key);
executeJob(jobKey);
} else {
throw new JobNotFoundException(String.valueOf(jobId));
}
} catch (JobNotFoundException e) {
throw new JobNotFoundException(String.valueOf(jobId));
}
}
private void scheduleJob(ScheduledJobDetail scheduledJobDetails) {
try {
JobDetail jobDetail = createJobDetail(scheduledJobDetails);
Trigger trigger = createTrigger(scheduledJobDetails, jobDetail);
schedulerFactoryBean.getScheduler().scheduleJob(jobDetail, trigger);
scheduledJobDetails.updateJobKey(getJobKeyAsString(jobDetail.getKey()));
scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime());
scheduledJobDetails.updateCurrentlyRunningStatus(false);
scheduledJobDetails.updateErrorLog(null);
} catch (Throwable throwable) {
scheduledJobDetails.updateNextRunTime(null);
String stackTrace = getStackTraceAsString(throwable);
scheduledJobDetails.updateErrorLog(stackTrace);
}
}
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;
}
}
MethodInvokingJobDetailFactoryBean jobDetailFactoryBean = new MethodInvokingJobDetailFactoryBean();
jobDetailFactoryBean.setName(scheduledJobDetails.getJobName() + "JobDetail" + tenant.getId());
jobDetailFactoryBean.setTargetObject(targetObject);
jobDetailFactoryBean.setTargetMethod(jobDetails[CronMethodParser.METHOD_INDEX]);
jobDetailFactoryBean.setGroup(scheduledJobDetails.getGroupName());
jobDetailFactoryBean.setConcurrent(false);
jobDetailFactoryBean.afterPropertiesSet();
return jobDetailFactoryBean.getObject();
}
/**
* @param scheduledJobDetails
* @return
*/
private Trigger createTrigger(ScheduledJobDetail scheduledJobDetails, JobDetail jobDetail) {
MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant();
CronTriggerFactoryBean cronTriggerFactoryBean = new CronTriggerFactoryBean();
cronTriggerFactoryBean.setName(scheduledJobDetails.getJobName() + "Trigger" + tenant.getId());
cronTriggerFactoryBean.setJobDetail(jobDetail);
JobDataMap jobDataMap = new JobDataMap();
jobDataMap.put(SchedularServiceConstants.TENANT_IDENTIFIER, tenant.getTenantIdentifier());
cronTriggerFactoryBean.setJobDataMap(jobDataMap);
TimeZone timeZone = TimeZone.getTimeZone(tenant.getTimezoneId());
cronTriggerFactoryBean.setTimeZone(timeZone);
cronTriggerFactoryBean.setGroup(scheduledJobDetails.getGroupName());
cronTriggerFactoryBean.setCronExpression(scheduledJobDetails.getCroneExpression());
cronTriggerFactoryBean.setPriority(scheduledJobDetails.getTaskPriority());
cronTriggerFactoryBean.afterPropertiesSet();
return cronTriggerFactoryBean.getObject();
}
/**
* @param throwable
* @return
*/
private String getStackTraceAsString(Throwable throwable) {
StackTraceElement[] stackTraceElements = throwable.getStackTrace();
StringBuffer sb = new StringBuffer(throwable.toString());
for (StackTraceElement element : stackTraceElements) {
sb.append("\n \t at ").append(element.getClassName()).append(".").append(element.getMethodName()).append("(")
.append(element.getLineNumber()).append(")");
}
return sb.toString();
}
private String getJobKeyAsString(JobKey jobKey) {
return jobKey.getName() + SchedularServiceConstants.JOB_KEY_SEPERATOR + jobKey.getGroup();
}
private JobKey constructJobKey(String Key) {
String[] keyParams = Key.split(SchedularServiceConstants.JOB_KEY_SEPERATOR);
JobKey JobKey = new JobKey(keyParams[0], keyParams[1]);
return JobKey;
}
}
@@ -7,14 +7,16 @@ import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobRunHistory;
public interface SchedularService {
public List<ScheduledJobDetail> getScheduledJobDetails();
public List<ScheduledJobDetail> retrieveAllJobs();
public ScheduledJobDetail getByTriggerKey(String triggerKey);
public ScheduledJobDetail findByJobKey(String triggerKey);
public void saveOrUpdate(ScheduledJobDetail scheduledJobDetails);
public void saveOrUpdate(ScheduledJobDetail scheduledJobDetails, ScheduledJobRunHistory scheduledJobRunHistory);
public Long getMaxVersionBy(String triggerKey);
public Long fetchMaxVersionBy(String triggerKey);
public ScheduledJobDetail findByJobId(Long jobId);
}
@@ -2,7 +2,7 @@ package org.mifosplatform.infrastructure.jobs.service;
public interface SchedularServiceConstants {
public static final String TRIGGER_KEY_SEPERATOR = " _ ";
public static final String JOB_KEY_SEPERATOR = " _ ";
public static final String TRIGGER_TYPE_CRON = "cron";
public static final String TRIGGER_TYPE_APPLICATION = "application";
public static final String TRIGGER_TYPE_REFERENCE = "TRIGGER_TYPE_REFERENCE";
@@ -25,13 +25,13 @@ public class SchedularServiceJpaRepositoryImpl implements SchedularService {
}
@Override
public List<ScheduledJobDetail> getScheduledJobDetails() {
public List<ScheduledJobDetail> retrieveAllJobs() {
return scheduledJobDetailsRepository.findAll();
}
@Override
public ScheduledJobDetail getByTriggerKey(final String triggerKey) {
return scheduledJobDetailsRepository.findByTriggerKey(triggerKey);
public ScheduledJobDetail findByJobKey(final String jobKey) {
return scheduledJobDetailsRepository.findByJobKey(jobKey);
}
@Transactional
@@ -48,12 +48,18 @@ public class SchedularServiceJpaRepositoryImpl implements SchedularService {
}
@Override
public Long getMaxVersionBy(final String triggerKey) {
public Long fetchMaxVersionBy(final String jobKey) {
Long version = 0L;
Long versionFromDB = scheduledJobRunHistoryRepository.findMaxVersionByTriggerKey(triggerKey);
Long versionFromDB = this.scheduledJobRunHistoryRepository.findMaxVersionByJobKey(jobKey);
if (versionFromDB != null) {
version = versionFromDB;
}
return version;
}
@Override
public ScheduledJobDetail findByJobId(Long jobId) {
return this.scheduledJobDetailsRepository.findByJobId(jobId);
}
}
@@ -6,12 +6,13 @@ 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;
import org.quartz.JobListener;
import org.quartz.Trigger;
import org.quartz.TriggerKey;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@@ -48,6 +49,13 @@ public class SchedulerJobListener implements JobListener {
String tenantIdentifier = context.getTrigger().getJobDataMap().getString(SchedularServiceConstants.TENANT_IDENTIFIER);
MifosPlatformTenant tenant = this.tenantDetailsService.loadTenantById(tenantIdentifier);
ThreadLocalContextUtil.setTenant(tenant);
JobKey key = context.getTrigger().getJobKey();
String jobKey = key.getName() + SchedularServiceConstants.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);
}
@Override
@@ -58,10 +66,10 @@ public class SchedulerJobListener implements JobListener {
@Override
public void jobWasExecuted(final JobExecutionContext context, final JobExecutionException jobException) {
Trigger trigger = context.getTrigger();
TriggerKey key = trigger.getKey();
String triggerKey = key.getName() + SchedularServiceConstants.TRIGGER_KEY_SEPERATOR + key.getGroup();
final ScheduledJobDetail scheduledJobDetails = schedularService.getByTriggerKey(triggerKey);
Long version = schedularService.getMaxVersionBy(triggerKey) + 1;
JobKey key = context.getJobDetail().getKey();
String jobKey = key.getName() + SchedularServiceConstants.JOB_KEY_SEPERATOR + key.getGroup();
final ScheduledJobDetail scheduledJobDetails = schedularService.findByJobKey(jobKey);
Long version = schedularService.fetchMaxVersionBy(jobKey) + 1;
String status = SchedularServiceConstants.STATUS_SUCCESS;
String errorMessage = null;
String errorLog = null;
@@ -84,11 +92,13 @@ public class SchedulerJobListener implements JobListener {
String triggerType = SchedularServiceConstants.TRIGGER_TYPE_CRON;
if (context.getMergedJobDataMap().containsKey(SchedularServiceConstants.TRIGGER_TYPE_REFERENCE)) {
triggerType = context.getMergedJobDataMap().getString(SchedularServiceConstants.TRIGGER_TYPE_REFERENCE);
}
if (triggerType == SchedularServiceConstants.TRIGGER_TYPE_CRON) {
scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime());
}
scheduledJobDetails.updatePreviousRunStartTime(context.getFireTime());
scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime());
scheduledJobDetails.updateCurrentlyRunningStatus(false);
ScheduledJobRunHistory runHistory = new ScheduledJobRunHistory(scheduledJobDetails, version, context.getFireTime(), new Date(),
status, errorMessage, triggerType, errorLog);
@@ -0,0 +1,18 @@
package org.mifosplatform.infrastructure.jobs.service;
import java.util.List;
import org.mifosplatform.infrastructure.core.service.Page;
import org.mifosplatform.infrastructure.jobs.data.JobDetailData;
import org.mifosplatform.infrastructure.jobs.data.JobDetailHistoryData;
public interface SchedulerJobRunnerReadService {
public List<JobDetailData> findAllJobDeatils();
public JobDetailData retrieveOne(Long jobId);
public Page<JobDetailHistoryData> retrieveJobHistory(Long jobId);
}
@@ -0,0 +1,124 @@
package org.mifosplatform.infrastructure.jobs.service;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Date;
import java.util.List;
import org.mifosplatform.infrastructure.core.service.Page;
import org.mifosplatform.infrastructure.core.service.PaginationHelper;
import org.mifosplatform.infrastructure.core.service.RoutingDataSource;
import org.mifosplatform.infrastructure.jobs.data.JobDetailData;
import org.mifosplatform.infrastructure.jobs.data.JobDetailHistoryData;
import org.mifosplatform.infrastructure.jobs.exception.JobNotFoundException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.EmptyResultDataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.stereotype.Service;
@Service
public class SchedulerJobRunnerReadServiceImpl implements SchedulerJobRunnerReadService {
private final JdbcTemplate jdbcTemplate;
private final PaginationHelper<JobDetailHistoryData> paginationHelper = new PaginationHelper<JobDetailHistoryData>();
@Autowired
public SchedulerJobRunnerReadServiceImpl(final RoutingDataSource dataSource) {
jdbcTemplate = new JdbcTemplate(dataSource);
}
@Override
public List<JobDetailData> findAllJobDeatils() {
JobDetailMapper detailMapper = new JobDetailMapper();
String sql = detailMapper.schema();
List<JobDetailData> JobDeatils = jdbcTemplate.query(sql, detailMapper, new Object[] {});
return JobDeatils;
}
@Override
public JobDetailData retrieveOne(Long jobId) {
try {
JobDetailMapper detailMapper = new JobDetailMapper();
String sql = detailMapper.schema() + " where job.id=?";
return jdbcTemplate.queryForObject(sql, detailMapper, new Object[] { jobId });
} catch (EmptyResultDataAccessException e) {
throw new JobNotFoundException(String.valueOf(jobId));
}
}
@Override
public Page<JobDetailHistoryData> retrieveJobHistory(Long jobId) {
retrieveOne(jobId);
JobHistoryMapper jobHistoryMapper = new JobHistoryMapper();
String sql = jobHistoryMapper.schema() + " where job.id=?";
final String sqlCountRows = "SELECT FOUND_ROWS()";
return paginationHelper.fetchPage(jdbcTemplate, sqlCountRows, sql, new Object[] { jobId }, jobHistoryMapper);
}
private static final class JobDetailMapper implements RowMapper<JobDetailData> {
private StringBuilder sqlBuilder = new StringBuilder("select")
.append(" job.id,job.display_name as displayName,job.next_run_time as nextRunTime,job.initializing_errorlog as initializingError,job.is_active as active,job.currently_running as currentlyRunning,")
.append(" runHistory.version,runHistory.start_time as lastRunStartTime,runHistory.end_time as lastRunEndTime,runHistory.`status`,runHistory.error_message as jobRunErrorMessage,runHistory.trigger_type as triggerType,runHistory.error_log as jobRunErrorLog ")
.append(" from job job left join job_run_history runHistory ON job.id=runHistory.job_id and job.previous_run_start_time=runHistory.start_time ");
public String schema() {
return sqlBuilder.toString();
}
@Override
public JobDetailData mapRow(ResultSet rs, @SuppressWarnings("unused") int rowNum) throws SQLException {
Long id = rs.getLong("id");
String displayName = rs.getString("displayName");
Date nextRunTime = rs.getDate("nextRunTime");
String initializingError = rs.getString("initializingError");
boolean active = rs.getBoolean("active");
boolean currentlyRunning = rs.getBoolean("currentlyRunning");
Long version = rs.getLong("version");
Date jobRunStartTime = rs.getDate("lastRunStartTime");
Date jobRunEndTime = rs.getDate("lastRunEndTime");
String status = rs.getString("status");
String jobRunErrorMessage = rs.getString("jobRunErrorMessage");
String triggerType = rs.getString("triggerType");
String jobRunErrorLog = rs.getString("jobRunErrorLog");
JobDetailHistoryData lastRunHistory = new JobDetailHistoryData(version, jobRunStartTime, jobRunEndTime, status,
jobRunErrorMessage, triggerType, jobRunErrorLog);
JobDetailData jobDetail = new JobDetailData(id, displayName, nextRunTime, initializingError, active, currentlyRunning,
lastRunHistory);
return jobDetail;
}
}
private static final class JobHistoryMapper implements RowMapper<JobDetailHistoryData> {
private StringBuilder sqlBuilder = new StringBuilder("select")
.append(" runHistory.version,runHistory.start_time as runStartTime,runHistory.end_time as runEndTime,runHistory.`status`,runHistory.error_message as jobRunErrorMessage,runHistory.trigger_type as triggerType,runHistory.error_log as jobRunErrorLog ")
.append(" from job job join job_run_history runHistory ON job.id=runHistory.job_id");
public String schema() {
return sqlBuilder.toString();
}
@Override
public JobDetailHistoryData mapRow(ResultSet rs, @SuppressWarnings("unused") int rowNum) throws SQLException {
Long version = rs.getLong("version");
Date jobRunStartTime = rs.getDate("runStartTime");
Date jobRunEndTime = rs.getDate("runEndTime");
String status = rs.getString("status");
String jobRunErrorMessage = rs.getString("jobRunErrorMessage");
String triggerType = rs.getString("triggerType");
String jobRunErrorLog = rs.getString("jobRunErrorLog");
JobDetailHistoryData jobDetailHistory = new JobDetailHistoryData(version, jobRunStartTime, jobRunEndTime, status,
jobRunErrorMessage, triggerType, jobRunErrorLog);
return jobDetailHistory;
}
}
}