added scheduler stop API

This commit is contained in:
Pramod Nuthakki
2013-07-25 19:09:54 +05:30
parent cb57df866c
commit ce95c68aa2
9 changed files with 273 additions and 97 deletions
@@ -0,0 +1,71 @@
package org.mifosplatform.infrastructure.jobs.api;
import javax.ws.rs.Consumes;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
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.apache.commons.lang.StringUtils;
import org.mifosplatform.infrastructure.core.api.ApiRequestParameterHelper;
import org.mifosplatform.infrastructure.core.exception.UnrecognizedQueryParamException;
import org.mifosplatform.infrastructure.core.serialization.ApiRequestJsonSerializationSettings;
import org.mifosplatform.infrastructure.core.serialization.ToApiJsonSerializer;
import org.mifosplatform.infrastructure.jobs.data.SchedulerDetailData;
import org.mifosplatform.infrastructure.jobs.service.JobRegisterService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Path("/scheduler")
@Component
public class SchedulerApiResource {
private final JobRegisterService jobRegisterService;
private final ToApiJsonSerializer<SchedulerDetailData> toApiJsonSerializer;
private final ApiRequestParameterHelper apiRequestParameterHelper;
@Autowired
public SchedulerApiResource(final JobRegisterService jobRegisterService,
final ToApiJsonSerializer<SchedulerDetailData> toApiJsonSerializer, final ApiRequestParameterHelper apiRequestParameterHelper) {
this.jobRegisterService = jobRegisterService;
this.toApiJsonSerializer = toApiJsonSerializer;
this.apiRequestParameterHelper = apiRequestParameterHelper;
}
@GET
@Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON })
public String retrieveStatus(@Context final UriInfo uriInfo) {
boolean isSchedulerRunning = jobRegisterService.isSchedulerRunning();
final ApiRequestJsonSerializationSettings settings = apiRequestParameterHelper.process(uriInfo.getQueryParameters());
SchedulerDetailData schedulerDetailData = new SchedulerDetailData(isSchedulerRunning);
return this.toApiJsonSerializer.serialize(settings, schedulerDetailData,
SchedulerJobApiConstants.SCHEDULER_DETAIL_RESPONSE_DATA_PARAMETERS);
}
@POST
@Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON })
public Response changeSchedulerStatus(@QueryParam(SchedulerJobApiConstants.COMMAND) final String commandParam) {
Response response = Response.status(400).build();
if (is(commandParam, SchedulerJobApiConstants.COMMAND_START_SCHEDULER)) {
jobRegisterService.startScheduler();
response = Response.status(202).build();
} else if (is(commandParam, SchedulerJobApiConstants.COMMAND_STOP_SCHEDULER)) {
jobRegisterService.pauseScheduler();
response = Response.status(202).build();
} else {
throw new UnrecognizedQueryParamException(SchedulerJobApiConstants.COMMAND, commandParam);
}
return response;
}
private boolean is(final String commandParam, final String commandValue) {
return StringUtils.isNotBlank(commandParam) && commandParam.trim().equalsIgnoreCase(commandValue);
}
}
@@ -6,7 +6,6 @@ import java.util.Set;
public class SchedulerJobApiConstants { public class SchedulerJobApiConstants {
public static final String JOB_RESOURCE_NAME = "schedulerjob"; public static final String JOB_RESOURCE_NAME = "schedulerjob";
// response parameters // response parameters
public static final String jobIdentifierParamName = "jobId"; public static final String jobIdentifierParamName = "jobId";
@@ -17,7 +16,6 @@ public class SchedulerJobApiConstants {
public static final String currentlyRunningParamName = "currentlyRunning"; public static final String currentlyRunningParamName = "currentlyRunning";
public static final String lastRunHistoryObjParamName = "lastRunHistory"; public static final String lastRunHistoryObjParamName = "lastRunHistory";
public static final String versionParamName = "version"; public static final String versionParamName = "version";
public static final String jobRunStartTimeParamName = "jobRunStartTime"; public static final String jobRunStartTimeParamName = "jobRunStartTime";
public static final String jobRunEndTimeParamName = "jobRunEndTime"; public static final String jobRunEndTimeParamName = "jobRunEndTime";
@@ -25,16 +23,27 @@ public class SchedulerJobApiConstants {
public static final String jobRunErrorMessageParamName = "jobRunErrorMessage"; public static final String jobRunErrorMessageParamName = "jobRunErrorMessage";
public static final String triggerTypeParamName = "triggerType"; public static final String triggerTypeParamName = "triggerType";
public static final String jobRunErrorLogParamName = "jobRunErrorLog"; public static final String jobRunErrorLogParamName = "jobRunErrorLog";
public static final String cronExpressionParamName = "cronExpression"; public static final String cronExpressionParamName = "cronExpression";
public static final String schedulerStatusParamName = "active";
public static final Set<String> JOB_DETAIL_RESPONSE_DATA_PARAMETERS = new HashSet<String>(Arrays.asList(jobIdentifierParamName, public static final Set<String> JOB_DETAIL_RESPONSE_DATA_PARAMETERS = new HashSet<String>(Arrays.asList(jobIdentifierParamName,
displayNameParamName, nextRunTimeParamName, initializingErrorParamName, jobActiveStatusParamName,currentlyRunningParamName, lastRunHistoryObjParamName)); displayNameParamName, nextRunTimeParamName, initializingErrorParamName, jobActiveStatusParamName, currentlyRunningParamName,
lastRunHistoryObjParamName));
public static final Set<String> JOB_HISTORY_RESPONSE_DATA_PARAMETERS = new HashSet<String>(Arrays.asList(versionParamName, public static final Set<String> JOB_HISTORY_RESPONSE_DATA_PARAMETERS = new HashSet<String>(Arrays.asList(versionParamName,
jobRunStartTimeParamName, jobRunEndTimeParamName, statusParamName, jobRunErrorMessageParamName, triggerTypeParamName, jobRunStartTimeParamName, jobRunEndTimeParamName, statusParamName, jobRunErrorMessageParamName, triggerTypeParamName,
jobRunErrorLogParamName)); jobRunErrorLogParamName));
public static final Set<String> JOB_UPDATE_REQUEST_DATA_PARAMETERS = new HashSet<String>(Arrays.asList(displayNameParamName,jobActiveStatusParamName,cronExpressionParamName)); public static final Set<String> JOB_UPDATE_REQUEST_DATA_PARAMETERS = new HashSet<String>(Arrays.asList(displayNameParamName,
jobActiveStatusParamName, cronExpressionParamName));
public static final Set<String> SCHEDULER_DETAIL_RESPONSE_DATA_PARAMETERS = new HashSet<String>(Arrays.asList(schedulerStatusParamName));
public static final String COMMAND_EXECUTE_JOB = "executeJob";
public static final String COMMAND_STOP_SCHEDULER = "stop";
public static final String COMMAND_START_SCHEDULER = "start";
public static final String COMMAND = "command";
public static final String JOB_ID = "jobId";
public static final String JOB_RUN_HISTORY = "runhistory";
public static final String SCHEDULER_STATUS_PATH = "scheduler";
} }
@@ -15,11 +15,13 @@ import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response; import javax.ws.rs.core.Response;
import javax.ws.rs.core.UriInfo; import javax.ws.rs.core.UriInfo;
import org.apache.commons.lang.StringUtils;
import org.mifosplatform.commands.domain.CommandWrapper; import org.mifosplatform.commands.domain.CommandWrapper;
import org.mifosplatform.commands.service.CommandWrapperBuilder; import org.mifosplatform.commands.service.CommandWrapperBuilder;
import org.mifosplatform.commands.service.PortfolioCommandSourceWritePlatformService; import org.mifosplatform.commands.service.PortfolioCommandSourceWritePlatformService;
import org.mifosplatform.infrastructure.core.api.ApiRequestParameterHelper; import org.mifosplatform.infrastructure.core.api.ApiRequestParameterHelper;
import org.mifosplatform.infrastructure.core.data.CommandProcessingResult; import org.mifosplatform.infrastructure.core.data.CommandProcessingResult;
import org.mifosplatform.infrastructure.core.exception.UnrecognizedQueryParamException;
import org.mifosplatform.infrastructure.core.serialization.ApiRequestJsonSerializationSettings; import org.mifosplatform.infrastructure.core.serialization.ApiRequestJsonSerializationSettings;
import org.mifosplatform.infrastructure.core.serialization.ToApiJsonSerializer; import org.mifosplatform.infrastructure.core.serialization.ToApiJsonSerializer;
import org.mifosplatform.infrastructure.core.service.Page; import org.mifosplatform.infrastructure.core.service.Page;
@@ -66,52 +68,50 @@ public class SchedulerJobApiResource {
} }
@GET @GET
@Path("{jobId}") @Path("{" + SchedulerJobApiConstants.JOB_ID + "}")
@Consumes({ MediaType.APPLICATION_JSON }) @Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON }) @Produces({ MediaType.APPLICATION_JSON })
public String retrieveOne(@PathParam("jobId") final Long jobId, @Context final UriInfo uriInfo) { public String retrieveOne(@PathParam(SchedulerJobApiConstants.JOB_ID) final Long jobId, @Context final UriInfo uriInfo) {
JobDetailData jobDetailData = this.schedulerJobRunnerReadService.retrieveOne(jobId); JobDetailData jobDetailData = this.schedulerJobRunnerReadService.retrieveOne(jobId);
final ApiRequestJsonSerializationSettings settings = apiRequestParameterHelper.process(uriInfo.getQueryParameters()); final ApiRequestJsonSerializationSettings settings = this.apiRequestParameterHelper.process(uriInfo.getQueryParameters());
return this.toApiJsonSerializer.serialize(settings, jobDetailData, SchedulerJobApiConstants.JOB_DETAIL_RESPONSE_DATA_PARAMETERS); return this.toApiJsonSerializer.serialize(settings, jobDetailData, SchedulerJobApiConstants.JOB_DETAIL_RESPONSE_DATA_PARAMETERS);
} }
@GET @GET
@Path("{jobId}/runhistory") @Path("{" + SchedulerJobApiConstants.JOB_ID + "}/" + SchedulerJobApiConstants.JOB_RUN_HISTORY)
@Consumes({ MediaType.APPLICATION_JSON }) @Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON }) @Produces({ MediaType.APPLICATION_JSON })
public String retrieveHistory(@Context final UriInfo uriInfo, @PathParam("jobId") final Long jobId, public String retrieveHistory(@Context final UriInfo uriInfo, @PathParam(SchedulerJobApiConstants.JOB_ID) final Long jobId,
@QueryParam("offset") final Integer offset, @QueryParam("limit") final Integer limit, @QueryParam("offset") final Integer offset, @QueryParam("limit") final Integer limit,
@QueryParam("orderBy") final String orderBy, @QueryParam("sortOrder") final String sortOrder) { @QueryParam("orderBy") final String orderBy, @QueryParam("sortOrder") final String sortOrder) {
final SearchParameters searchParameters = SearchParameters.forPagination(offset, limit, orderBy, sortOrder); final SearchParameters searchParameters = SearchParameters.forPagination(offset, limit, orderBy, sortOrder);
Page<JobDetailHistoryData> jobhistoryDetailData = this.schedulerJobRunnerReadService.retrieveJobHistory(jobId, searchParameters); Page<JobDetailHistoryData> jobhistoryDetailData = this.schedulerJobRunnerReadService.retrieveJobHistory(jobId, searchParameters);
final ApiRequestJsonSerializationSettings settings = apiRequestParameterHelper.process(uriInfo.getQueryParameters()); final ApiRequestJsonSerializationSettings settings = this.apiRequestParameterHelper.process(uriInfo.getQueryParameters());
return this.jobHistoryToApiJsonSerializer.serialize(settings, jobhistoryDetailData, return this.jobHistoryToApiJsonSerializer.serialize(settings, jobhistoryDetailData,
SchedulerJobApiConstants.JOB_HISTORY_RESPONSE_DATA_PARAMETERS); SchedulerJobApiConstants.JOB_HISTORY_RESPONSE_DATA_PARAMETERS);
} }
@POST @POST
@Path("{jobId}/executeJob") @Path("{" + SchedulerJobApiConstants.JOB_ID + "}")
@Consumes({ MediaType.APPLICATION_JSON }) @Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON }) @Produces({ MediaType.APPLICATION_JSON })
public Response executeJob(@PathParam("jobId") final Long jobId) { public Response executeJob(@PathParam(SchedulerJobApiConstants.JOB_ID) final Long jobId,
/** @QueryParam(SchedulerJobApiConstants.COMMAND) final String commandParam) {
* TODO : this API needs to be refactored to accept a mandatory command
* param i.e POST to jobs/{jobId}?command=run would trigger the job
* execution, any other command or no command would throw and invalid
* command exception..similar to
* https://demo.openmf.org/api-docs/apiLive.htm#loans_approve
**/
Response response = Response.status(400).build(); Response response = Response.status(400).build();
jobRegisterService.executeJob(jobId); if (is(commandParam, SchedulerJobApiConstants.COMMAND_EXECUTE_JOB)) {
response = Response.status(202).build(); this.jobRegisterService.executeJob(jobId);
response = Response.status(202).build();
} else {
throw new UnrecognizedQueryParamException(SchedulerJobApiConstants.COMMAND, commandParam);
}
return response; return response;
} }
@PUT @PUT
@Path("{jobId}") @Path("{" + SchedulerJobApiConstants.JOB_ID + "}")
@Consumes({ MediaType.APPLICATION_JSON }) @Consumes({ MediaType.APPLICATION_JSON })
@Produces({ MediaType.APPLICATION_JSON }) @Produces({ MediaType.APPLICATION_JSON })
public String updateJobDetail(@PathParam("jobId") final Long jobId, final String jsonRequestBody) { public String updateJobDetail(@PathParam(SchedulerJobApiConstants.JOB_ID) final Long jobId, final String jsonRequestBody) {
final CommandWrapper commandRequest = new CommandWrapperBuilder() // final CommandWrapper commandRequest = new CommandWrapperBuilder() //
.updateJobDetail(jobId) // .updateJobDetail(jobId) //
@@ -125,4 +125,9 @@ public class SchedulerJobApiResource {
} }
return this.toApiJsonSerializer.serialize(result); return this.toApiJsonSerializer.serialize(result);
} }
private boolean is(final String commandParam, final String commandValue) {
return StringUtils.isNotBlank(commandParam) && commandParam.trim().equalsIgnoreCase(commandValue);
}
} }
@@ -0,0 +1,11 @@
package org.mifosplatform.infrastructure.jobs.data;
public class SchedulerDetailData {
@SuppressWarnings("unused")
private final boolean active;
public SchedulerDetailData(final boolean active) {
this.active = active;
}
}
@@ -1,15 +1,18 @@
package org.mifosplatform.infrastructure.jobs.service; package org.mifosplatform.infrastructure.jobs.service;
public interface JobRegisterService { public interface JobRegisterService {
public void executeJob(Long jobId); public void executeJob(Long jobId);
public void rescheduleJob(Long jobId); public void rescheduleJob(Long jobId);
public void stopScheduler(); public void pauseScheduler();
public void startScheduler(); public void startScheduler();
public boolean isSchedulerRunning(); public boolean isSchedulerRunning();
public void stopScheduler(String name);
} }
@@ -1,5 +1,6 @@
package org.mifosplatform.infrastructure.jobs.service; package org.mifosplatform.infrastructure.jobs.service;
import java.util.ArrayList;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
@@ -9,6 +10,7 @@ import java.util.TimeZone;
import javax.annotation.PostConstruct; import javax.annotation.PostConstruct;
import org.mifosplatform.infrastructure.core.domain.MifosPlatformTenant; import org.mifosplatform.infrastructure.core.domain.MifosPlatformTenant;
import org.mifosplatform.infrastructure.core.exception.PlatformInternalServerException;
import org.mifosplatform.infrastructure.core.service.ThreadLocalContextUtil; import org.mifosplatform.infrastructure.core.service.ThreadLocalContextUtil;
import org.mifosplatform.infrastructure.jobs.annotation.CronMethodParser; import org.mifosplatform.infrastructure.jobs.annotation.CronMethodParser;
import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobDetail; import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobDetail;
@@ -76,28 +78,35 @@ public class JobRegisterServiceImpl implements JobRegisterService {
public void executeJob(final ScheduledJobDetail scheduledJobDetail) { public void executeJob(final ScheduledJobDetail scheduledJobDetail) {
try { try {
final JobDataMap jobDataMap = new JobDataMap(); final JobDataMap jobDataMap = new JobDataMap();
jobDataMap.put(SchedularServiceConstants.TRIGGER_TYPE_REFERENCE, SchedularServiceConstants.TRIGGER_TYPE_APPLICATION); jobDataMap.put(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE, SchedulerServiceConstants.TRIGGER_TYPE_APPLICATION);
jobDataMap.put(SchedularServiceConstants.TENANT_IDENTIFIER, ThreadLocalContextUtil.getTenant().getTenantIdentifier()); jobDataMap.put(SchedulerServiceConstants.TENANT_IDENTIFIER, ThreadLocalContextUtil.getTenant().getTenantIdentifier());
String key = scheduledJobDetail.getJobKey(); String key = scheduledJobDetail.getJobKey();
JobKey jobKey = constructJobKey(key); JobKey jobKey = constructJobKey(key);
String schedulerName = getSchedulerName(scheduledJobDetail); String schedulerName = getSchedulerName(scheduledJobDetail);
Scheduler scheduler = schedulers.get(schedulerName); Scheduler scheduler = schedulers.get(schedulerName);
if (scheduler != null && scheduler.checkExists(jobKey)) { if (scheduler == null || scheduler.isInStandbyMode() || !scheduler.checkExists(jobKey)) {
try { JobDetail jobDetail = createJobDetail(scheduledJobDetail);
scheduler.triggerJob(jobKey, jobDataMap); String tempSchedulerName = "temp" + scheduledJobDetail.getId();
} catch (SchedulerException e) { List<Class<? extends JobListener>> listenerClasses = new ArrayList<Class<? extends JobListener>>(2);
logger.error(e.getMessage(), e); listenerClasses.add(SchedulerJobListener.class);
} listenerClasses.add(SchedulerStopListener.class);
Scheduler tempScheduler = createScheduler(tempSchedulerName, 1, listenerClasses);
tempScheduler.addJob(jobDetail, true);
jobDataMap.put(SchedulerServiceConstants.SCHEDULER_NAME, tempSchedulerName);
this.schedulers.put(tempSchedulerName, tempScheduler);
tempScheduler.triggerJob(jobDetail.getKey(), jobDataMap);
} else { } else {
throw new JobNotFoundException(jobKey.toString()); scheduler.triggerJob(jobKey, jobDataMap);
} }
} catch (SchedulerException e) {
logger.error(e.getMessage(), e); } catch (Exception e) {
throw new PlatformInternalServerException("error.msg.sheduler.job.execution.failed", "Job execution failed for job with id:"
+ scheduledJobDetail.getId(), scheduledJobDetail.getId());
} }
} }
public void rescheduleJob(ScheduledJobDetail scheduledJobDetail) { public void rescheduleJob(final ScheduledJobDetail scheduledJobDetail) {
try { try {
String jobIdentity = scheduledJobDetail.getJobKey(); String jobIdentity = scheduledJobDetail.getJobKey();
JobKey jobKey = constructJobKey(jobIdentity); JobKey jobKey = constructJobKey(jobIdentity);
@@ -116,7 +125,7 @@ public class JobRegisterServiceImpl implements JobRegisterService {
} }
@Override @Override
public void stopScheduler() { public void pauseScheduler() {
schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), false); schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), false);
for (Map.Entry<String, Scheduler> schedulerEntry : this.schedulers.entrySet()) { for (Map.Entry<String, Scheduler> schedulerEntry : this.schedulers.entrySet()) {
try { try {
@@ -132,7 +141,6 @@ public class JobRegisterServiceImpl implements JobRegisterService {
@Override @Override
public void startScheduler() { public void startScheduler() {
schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), true);
for (Map.Entry<String, Scheduler> schedulerEntry : this.schedulers.entrySet()) { for (Map.Entry<String, Scheduler> schedulerEntry : this.schedulers.entrySet()) {
try { try {
String schedulerName = schedulerEntry.getValue().getSchedulerName(); String schedulerName = schedulerEntry.getValue().getSchedulerName();
@@ -143,26 +151,20 @@ public class JobRegisterServiceImpl implements JobRegisterService {
logger.error(e.getMessage(), e); logger.error(e.getMessage(), e);
} }
} }
schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), true);
} }
@Override @Override
public void rescheduleJob(Long jobId) { public void rescheduleJob(final Long jobId) {
ScheduledJobDetail scheduledJobDetail = this.schedularWritePlatformService.findByJobId(jobId); ScheduledJobDetail scheduledJobDetail = this.schedularWritePlatformService.findByJobId(jobId);
rescheduleJob(scheduledJobDetail); rescheduleJob(scheduledJobDetail);
} }
@Override @Override
public void executeJob(Long jobId) { public void executeJob(final Long jobId) {
try { ScheduledJobDetail scheduledJobDetail = this.schedularWritePlatformService.findByJobId(jobId);
ScheduledJobDetail scheduledJobDetail = this.schedularWritePlatformService.findByJobId(jobId); if (scheduledJobDetail == null) { throw new JobNotFoundException(String.valueOf(jobId)); }
if (scheduledJobDetail != null) { executeJob(scheduledJobDetail);
executeJob(scheduledJobDetail);
} else {
throw new JobNotFoundException(String.valueOf(jobId));
}
} catch (JobNotFoundException e) {
throw new JobNotFoundException(String.valueOf(jobId));
}
} }
@Override @Override
@@ -170,27 +172,27 @@ public class JobRegisterServiceImpl implements JobRegisterService {
return this.schedulerStatus.get(ThreadLocalContextUtil.getTenant().getId()); return this.schedulerStatus.get(ThreadLocalContextUtil.getTenant().getId());
} }
private boolean isCurrentTenantsScheduler(String schedulerName) { private boolean isCurrentTenantsScheduler(final String schedulerName) {
int beginIndex = SchedularServiceConstants.SCHEDULER.length(); int beginIndex = SchedulerServiceConstants.SCHEDULER.length();
int endIndex = schedulerName.length(); int endIndex = schedulerName.length();
if (schedulerName.contains(SchedularServiceConstants.SCHEDULER_GROUP)) { if (schedulerName.contains(SchedulerServiceConstants.SCHEDULER_GROUP)) {
endIndex = schedulerName.indexOf(SchedularServiceConstants.SCHEDULER_GROUP); endIndex = schedulerName.indexOf(SchedulerServiceConstants.SCHEDULER_GROUP);
} }
String tenantId = schedulerName.substring(beginIndex, endIndex); String tenantId = schedulerName.substring(beginIndex, endIndex);
MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant(); MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant();
return tenant.getId() == Long.valueOf(tenantId); return tenant.getId() == Long.valueOf(tenantId);
} }
private void scheduleJob(ScheduledJobDetail scheduledJobDetails) { private void scheduleJob(final ScheduledJobDetail scheduledJobDetails) {
if (!scheduledJobDetails.isActiveSchedular()) { if (!scheduledJobDetails.isActiveSchedular()) {
scheduledJobDetails.updateNextRunTime(null); scheduledJobDetails.updateNextRunTime(null);
scheduledJobDetails.updateCurrentlyRunningStatus(false); scheduledJobDetails.updateCurrentlyRunningStatus(false);
return; return;
} }
try { try {
Scheduler scheduler = getScheduler(scheduledJobDetails);
JobDetail jobDetail = createJobDetail(scheduledJobDetails); JobDetail jobDetail = createJobDetail(scheduledJobDetails);
Trigger trigger = createTrigger(scheduledJobDetails, jobDetail); Trigger trigger = createTrigger(scheduledJobDetails, jobDetail);
Scheduler scheduler = getScheduler(scheduledJobDetails);
scheduler.scheduleJob(jobDetail, trigger); scheduler.scheduleJob(jobDetail, trigger);
scheduledJobDetails.updateJobKey(getJobKeyAsString(jobDetail.getKey())); scheduledJobDetails.updateJobKey(getJobKeyAsString(jobDetail.getKey()));
scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime()); scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime());
@@ -203,34 +205,49 @@ public class JobRegisterServiceImpl implements JobRegisterService {
scheduledJobDetails.updateCurrentlyRunningStatus(false); scheduledJobDetails.updateCurrentlyRunningStatus(false);
} }
private Scheduler getScheduler(ScheduledJobDetail scheduledJobDetail) throws Exception { private Scheduler getScheduler(final ScheduledJobDetail scheduledJobDetail) throws Exception {
String schedulername = getSchedulerName(scheduledJobDetail); String schedulername = getSchedulerName(scheduledJobDetail);
Scheduler scheduler = this.schedulers.get(schedulername); Scheduler scheduler = this.schedulers.get(schedulername);
if (scheduler == null) { if (scheduler == null) {
int noOfThreads = SchedularServiceConstants.DEFAULT_THREAD_COUNT; int noOfThreads = SchedulerServiceConstants.DEFAULT_THREAD_COUNT;
if (scheduledJobDetail.getSchedulerGroup() > 0) { if (scheduledJobDetail.getSchedulerGroup() > 0) {
noOfThreads = SchedularServiceConstants.GROUP_THREAD_COUNT; noOfThreads = SchedulerServiceConstants.GROUP_THREAD_COUNT;
} }
scheduler = createScheduler(schedulername, noOfThreads); scheduler = createScheduler(schedulername, noOfThreads, null);
this.schedulers.put(schedulername, scheduler); this.schedulers.put(schedulername, scheduler);
} }
return scheduler; return scheduler;
} }
private String getSchedulerName(ScheduledJobDetail scheduledJobDetail) { @Override
public void stopScheduler(String name) {
Scheduler scheduler = this.schedulers.remove(name);
try {
scheduler.shutdown();
} catch (SchedulerException e) {
logger.error(e.getMessage(), e);
}
}
private String getSchedulerName(final ScheduledJobDetail scheduledJobDetail) {
StringBuilder sb = new StringBuilder(20); StringBuilder sb = new StringBuilder(20);
MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant(); MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant();
sb.append(SchedularServiceConstants.SCHEDULER).append(tenant.getId()); sb.append(SchedulerServiceConstants.SCHEDULER).append(tenant.getId());
if (scheduledJobDetail.getSchedulerGroup() > 0) { if (scheduledJobDetail.getSchedulerGroup() > 0) {
sb.append(SchedularServiceConstants.SCHEDULER_GROUP).append(scheduledJobDetail.getSchedulerGroup()); sb.append(SchedulerServiceConstants.SCHEDULER_GROUP).append(scheduledJobDetail.getSchedulerGroup());
} }
return sb.toString(); return sb.toString();
} }
private Scheduler createScheduler(String name, int noOfThreads) throws Exception { private Scheduler createScheduler(final String name, final int noOfThreads, List<Class<? extends JobListener>> listenerClasses)
throws Exception {
SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean(); SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();
schedulerFactoryBean.setSchedulerName(name); schedulerFactoryBean.setSchedulerName(name);
schedulerFactoryBean.setGlobalJobListeners(getGlobalListener()); if (listenerClasses == null) {
listenerClasses = new ArrayList<Class<? extends JobListener>>(1);
listenerClasses.add(SchedulerJobListener.class);
}
schedulerFactoryBean.setGlobalJobListeners(getGlobalListener(listenerClasses));
Properties quartzProperties = new Properties(); Properties quartzProperties = new Properties();
quartzProperties.put(SchedulerFactoryBean.PROP_THREAD_COUNT, Integer.toString(noOfThreads)); quartzProperties.put(SchedulerFactoryBean.PROP_THREAD_COUNT, Integer.toString(noOfThreads));
schedulerFactoryBean.setQuartzProperties(quartzProperties); schedulerFactoryBean.setQuartzProperties(quartzProperties);
@@ -239,21 +256,25 @@ public class JobRegisterServiceImpl implements JobRegisterService {
return schedulerFactoryBean.getScheduler(); return schedulerFactoryBean.getScheduler();
} }
private JobListener[] getGlobalListener() throws ClassNotFoundException { private JobListener[] getGlobalListener(List<Class<? extends JobListener>> listenerClasses) throws ClassNotFoundException {
JobListener listener = (JobListener) getBeanObject(SchedulerJobListener.class); List<JobListener> listeners = new ArrayList<JobListener>(listenerClasses.size());
JobListener[] listenerArray = { listener }; for (Class<?> listenerClass : listenerClasses) {
return listenerArray; JobListener listener = (JobListener) getBeanObject(listenerClass);
listeners.add(listener);
}
JobListener[] listenerArray = new JobListener[listeners.size()];
return listeners.toArray(listenerArray);
} }
private JobDetail createJobDetail(ScheduledJobDetail scheduledJobDetails) throws Exception { private JobDetail createJobDetail(final ScheduledJobDetail scheduledJobDetail) throws Exception {
MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant(); MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant();
String[] jobDetails = CronMethodParser.findTargetMethodDetails(scheduledJobDetails.getJobName()); String[] jobDetails = CronMethodParser.findTargetMethodDetails(scheduledJobDetail.getJobName());
Object targetObject = getBeanObject(Class.forName(jobDetails[CronMethodParser.CLASS_INDEX])); Object targetObject = getBeanObject(Class.forName(jobDetails[CronMethodParser.CLASS_INDEX]));
MethodInvokingJobDetailFactoryBean jobDetailFactoryBean = new MethodInvokingJobDetailFactoryBean(); MethodInvokingJobDetailFactoryBean jobDetailFactoryBean = new MethodInvokingJobDetailFactoryBean();
jobDetailFactoryBean.setName(scheduledJobDetails.getJobName() + "JobDetail" + tenant.getId()); jobDetailFactoryBean.setName(scheduledJobDetail.getJobName() + "JobDetail" + tenant.getId());
jobDetailFactoryBean.setTargetObject(targetObject); jobDetailFactoryBean.setTargetObject(targetObject);
jobDetailFactoryBean.setTargetMethod(jobDetails[CronMethodParser.METHOD_INDEX]); jobDetailFactoryBean.setTargetMethod(jobDetails[CronMethodParser.METHOD_INDEX]);
jobDetailFactoryBean.setGroup(scheduledJobDetails.getGroupName()); jobDetailFactoryBean.setGroup(scheduledJobDetail.getGroupName());
jobDetailFactoryBean.setConcurrent(false); jobDetailFactoryBean.setConcurrent(false);
jobDetailFactoryBean.afterPropertiesSet(); jobDetailFactoryBean.afterPropertiesSet();
return jobDetailFactoryBean.getObject(); return jobDetailFactoryBean.getObject();
@@ -280,13 +301,13 @@ public class JobRegisterServiceImpl implements JobRegisterService {
* @param scheduledJobDetails * @param scheduledJobDetails
* @return * @return
*/ */
private Trigger createTrigger(ScheduledJobDetail scheduledJobDetails, JobDetail jobDetail) { private Trigger createTrigger(final ScheduledJobDetail scheduledJobDetails, final JobDetail jobDetail) {
MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant(); MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant();
CronTriggerFactoryBean cronTriggerFactoryBean = new CronTriggerFactoryBean(); CronTriggerFactoryBean cronTriggerFactoryBean = new CronTriggerFactoryBean();
cronTriggerFactoryBean.setName(scheduledJobDetails.getJobName() + "Trigger" + tenant.getId()); cronTriggerFactoryBean.setName(scheduledJobDetails.getJobName() + "Trigger" + tenant.getId());
cronTriggerFactoryBean.setJobDetail(jobDetail); cronTriggerFactoryBean.setJobDetail(jobDetail);
JobDataMap jobDataMap = new JobDataMap(); JobDataMap jobDataMap = new JobDataMap();
jobDataMap.put(SchedularServiceConstants.TENANT_IDENTIFIER, tenant.getTenantIdentifier()); jobDataMap.put(SchedulerServiceConstants.TENANT_IDENTIFIER, tenant.getTenantIdentifier());
cronTriggerFactoryBean.setJobDataMap(jobDataMap); cronTriggerFactoryBean.setJobDataMap(jobDataMap);
TimeZone timeZone = TimeZone.getTimeZone(tenant.getTimezoneId()); TimeZone timeZone = TimeZone.getTimeZone(tenant.getTimezoneId());
cronTriggerFactoryBean.setTimeZone(timeZone); cronTriggerFactoryBean.setTimeZone(timeZone);
@@ -311,12 +332,12 @@ public class JobRegisterServiceImpl implements JobRegisterService {
return sb.toString(); return sb.toString();
} }
private String getJobKeyAsString(JobKey jobKey) { private String getJobKeyAsString(final JobKey jobKey) {
return jobKey.getName() + SchedularServiceConstants.JOB_KEY_SEPERATOR + jobKey.getGroup(); return jobKey.getName() + SchedulerServiceConstants.JOB_KEY_SEPERATOR + jobKey.getGroup();
} }
private JobKey constructJobKey(String Key) { private JobKey constructJobKey(final String Key) {
String[] keyParams = Key.split(SchedularServiceConstants.JOB_KEY_SEPERATOR); String[] keyParams = Key.split(SchedulerServiceConstants.JOB_KEY_SEPERATOR);
JobKey JobKey = new JobKey(keyParams[0], keyParams[1]); JobKey JobKey = new JobKey(keyParams[0], keyParams[1]);
return JobKey; return JobKey;
} }
@@ -27,7 +27,7 @@ public class SchedulerJobListener implements JobListener {
private int stackTraceLevel = 0; private int stackTraceLevel = 0;
private final String name = SchedularServiceConstants.DEFAULT_LISTENER_NAME; private final String name = SchedulerServiceConstants.DEFAULT_LISTENER_NAME;
private final SchedularWritePlatformService schedularService; private final SchedularWritePlatformService schedularService;
@@ -46,11 +46,11 @@ public class SchedulerJobListener implements JobListener {
@Override @Override
public void jobToBeExecuted(final JobExecutionContext context) { public void jobToBeExecuted(final JobExecutionContext context) {
String tenantIdentifier = context.getTrigger().getJobDataMap().getString(SchedularServiceConstants.TENANT_IDENTIFIER); String tenantIdentifier = context.getTrigger().getJobDataMap().getString(SchedulerServiceConstants.TENANT_IDENTIFIER);
MifosPlatformTenant tenant = this.tenantDetailsService.loadTenantById(tenantIdentifier); MifosPlatformTenant tenant = this.tenantDetailsService.loadTenantById(tenantIdentifier);
ThreadLocalContextUtil.setTenant(tenant); ThreadLocalContextUtil.setTenant(tenant);
JobKey key = context.getTrigger().getJobKey(); JobKey key = context.getTrigger().getJobKey();
String jobKey = key.getName() + SchedularServiceConstants.JOB_KEY_SEPERATOR + key.getGroup(); String jobKey = key.getName() + SchedulerServiceConstants.JOB_KEY_SEPERATOR + key.getGroup();
final ScheduledJobDetail scheduledJobDetail = schedularService.findByJobKey(jobKey); final ScheduledJobDetail scheduledJobDetail = schedularService.findByJobKey(jobKey);
if (scheduledJobDetail.isCurrentlyRunning()) { throw new JobInProcessExecution(scheduledJobDetail.getJobName()); } if (scheduledJobDetail.isCurrentlyRunning()) { throw new JobInProcessExecution(scheduledJobDetail.getJobName()); }
scheduledJobDetail.updateCurrentlyRunningStatus(true); scheduledJobDetail.updateCurrentlyRunningStatus(true);
@@ -67,14 +67,14 @@ public class SchedulerJobListener implements JobListener {
public void jobWasExecuted(final JobExecutionContext context, final JobExecutionException jobException) { public void jobWasExecuted(final JobExecutionContext context, final JobExecutionException jobException) {
Trigger trigger = context.getTrigger(); Trigger trigger = context.getTrigger();
JobKey key = context.getJobDetail().getKey(); JobKey key = context.getJobDetail().getKey();
String jobKey = key.getName() + SchedularServiceConstants.JOB_KEY_SEPERATOR + key.getGroup(); String jobKey = key.getName() + SchedulerServiceConstants.JOB_KEY_SEPERATOR + key.getGroup();
final ScheduledJobDetail scheduledJobDetails = schedularService.findByJobKey(jobKey); final ScheduledJobDetail scheduledJobDetails = schedularService.findByJobKey(jobKey);
Long version = schedularService.fetchMaxVersionBy(jobKey) + 1; Long version = schedularService.fetchMaxVersionBy(jobKey) + 1;
String status = SchedularServiceConstants.STATUS_SUCCESS; String status = SchedulerServiceConstants.STATUS_SUCCESS;
String errorMessage = null; String errorMessage = null;
String errorLog = null; String errorLog = null;
if (jobException != null) { if (jobException != null) {
status = SchedularServiceConstants.STATUS_FAILED; status = SchedulerServiceConstants.STATUS_FAILED;
stackTraceLevel = 0; stackTraceLevel = 0;
Throwable throwable = getCauseFromException(jobException); Throwable throwable = getCauseFromException(jobException);
stackTraceLevel = 0; stackTraceLevel = 0;
@@ -89,11 +89,11 @@ public class SchedulerJobListener implements JobListener {
errorLog = sb.toString(); errorLog = sb.toString();
} }
String triggerType = SchedularServiceConstants.TRIGGER_TYPE_CRON; String triggerType = SchedulerServiceConstants.TRIGGER_TYPE_CRON;
if (context.getMergedJobDataMap().containsKey(SchedularServiceConstants.TRIGGER_TYPE_REFERENCE)) { if (context.getMergedJobDataMap().containsKey(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE)) {
triggerType = context.getMergedJobDataMap().getString(SchedularServiceConstants.TRIGGER_TYPE_REFERENCE); triggerType = context.getMergedJobDataMap().getString(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE);
} }
if (triggerType == SchedularServiceConstants.TRIGGER_TYPE_CRON) { if (triggerType == SchedulerServiceConstants.TRIGGER_TYPE_CRON) {
scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime()); scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime());
} }
@@ -109,11 +109,11 @@ public class SchedulerJobListener implements JobListener {
} }
private Throwable getCauseFromException(final Throwable exception) { private Throwable getCauseFromException(final Throwable exception) {
if (stackTraceLevel <= SchedularServiceConstants.STACK_TRACE_LEVEL if (stackTraceLevel <= SchedulerServiceConstants.STACK_TRACE_LEVEL
&& exception.getCause() != null && exception.getCause() != null
&& (exception.getCause().toString().contains(SchedularServiceConstants.SCHEDULAR_EXCEPTION) && (exception.getCause().toString().contains(SchedulerServiceConstants.SCHEDULER_EXCEPTION)
|| exception.getCause().toString().contains(SchedularServiceConstants.JOB_EXECUTION_EXCEPTION) || exception || exception.getCause().toString().contains(SchedulerServiceConstants.JOB_EXECUTION_EXCEPTION) || exception
.getCause().toString().contains(SchedularServiceConstants.JOB_METHOD_INVOCATION_FAILED_EXCEPTION))) { .getCause().toString().contains(SchedulerServiceConstants.JOB_METHOD_INVOCATION_FAILED_EXCEPTION))) {
stackTraceLevel++; stackTraceLevel++;
return getCauseFromException(exception.getCause()); return getCauseFromException(exception.getCause());
} else if (exception.getCause() != null) { return exception.getCause(); } } else if (exception.getCause() != null) { return exception.getCause(); }
@@ -1,12 +1,12 @@
package org.mifosplatform.infrastructure.jobs.service; package org.mifosplatform.infrastructure.jobs.service;
public interface SchedularServiceConstants { public interface SchedulerServiceConstants {
public static final String JOB_KEY_SEPERATOR = " _ "; public static final String JOB_KEY_SEPERATOR = " _ ";
public static final String TRIGGER_TYPE_CRON = "cron"; public static final String TRIGGER_TYPE_CRON = "cron";
public static final String TRIGGER_TYPE_APPLICATION = "application"; public static final String TRIGGER_TYPE_APPLICATION = "application";
public static final String TRIGGER_TYPE_REFERENCE = "TRIGGER_TYPE_REFERENCE"; public static final String TRIGGER_TYPE_REFERENCE = "TRIGGER_TYPE_REFERENCE";
public static final String SCHEDULAR_EXCEPTION = "SchedulerException"; public static final String SCHEDULER_EXCEPTION = "SchedulerException";
public static final String JOB_EXECUTION_EXCEPTION = "JobExecutionException"; public static final String JOB_EXECUTION_EXCEPTION = "JobExecutionException";
public static final String JOB_METHOD_INVOCATION_FAILED_EXCEPTION = "JobMethodInvocationFailedException"; public static final String JOB_METHOD_INVOCATION_FAILED_EXCEPTION = "JobMethodInvocationFailedException";
public static final String STATUS_SUCCESS = "success"; public static final String STATUS_SUCCESS = "success";
@@ -18,5 +18,6 @@ public interface SchedularServiceConstants {
public static final String SCHEDULER_GROUP = "group"; public static final String SCHEDULER_GROUP = "group";
public static final int DEFAULT_THREAD_COUNT = 7; public static final int DEFAULT_THREAD_COUNT = 7;
public static final int GROUP_THREAD_COUNT = 1; public static final int GROUP_THREAD_COUNT = 1;
public static final String SCHEDULER_NAME = "schedulerName";
} }
@@ -0,0 +1,55 @@
package org.mifosplatform.infrastructure.jobs.service;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.JobListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
/**
* Global job Listener class to Stop the temporary scheduler once job execution
* completes
*/
@Component
public class SchedulerStopListener implements JobListener {
private static final String name = "Singlr Trigger Global Listner";
private final JobRegisterService jobRegisterService;
@Autowired
public SchedulerStopListener(final JobRegisterService jobRegisterService) {
this.jobRegisterService = jobRegisterService;
}
@Override
public String getName() {
return name;
}
@Override
public void jobToBeExecuted(@SuppressWarnings("unused") final JobExecutionContext context) {
}
@Override
public void jobExecutionVetoed(@SuppressWarnings("unused") final JobExecutionContext context) {
}
@Override
public void jobWasExecuted(final JobExecutionContext context, @SuppressWarnings("unused") final JobExecutionException jobException) {
final String schedulerName = context.getTrigger().getJobDataMap().getString(SchedulerServiceConstants.SCHEDULER_NAME);
if (schedulerName != null) {
Thread newThread = new Thread(new Runnable() {
@Override
public void run() {
jobRegisterService.stopScheduler(schedulerName);
}
});
newThread.run();
}
}
}