From ce95c68aa27e9189ebdccd704716bb14afaa22ae Mon Sep 17 00:00:00 2001 From: Pramod Nuthakki Date: Thu, 25 Jul 2013 15:36:29 +0530 Subject: [PATCH] added scheduler stop API --- .../jobs/api/SchedulerApiResource.java | 71 ++++++++++ .../jobs/api/SchedulerJobApiConstants.java | 21 ++- .../jobs/api/SchedulerJobApiResource.java | 43 +++--- .../jobs/data/SchedulerDetailData.java | 11 ++ .../jobs/service/JobRegisterService.java | 5 +- .../jobs/service/JobRegisterServiceImpl.java | 131 ++++++++++-------- .../jobs/service/SchedulerJobListener.java | 28 ++-- ...ts.java => SchedulerServiceConstants.java} | 5 +- .../jobs/service/SchedulerStopListener.java | 55 ++++++++ 9 files changed, 273 insertions(+), 97 deletions(-) create mode 100755 mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/api/SchedulerApiResource.java create mode 100755 mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/data/SchedulerDetailData.java rename mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/{SchedularServiceConstants.java => SchedulerServiceConstants.java} (86%) create mode 100755 mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerStopListener.java diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/api/SchedulerApiResource.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/api/SchedulerApiResource.java new file mode 100755 index 000000000..159dadedb --- /dev/null +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/api/SchedulerApiResource.java @@ -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 toApiJsonSerializer; + private final ApiRequestParameterHelper apiRequestParameterHelper; + + @Autowired + public SchedulerApiResource(final JobRegisterService jobRegisterService, + final ToApiJsonSerializer 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); + } +} diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/api/SchedulerJobApiConstants.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/api/SchedulerJobApiConstants.java index 8cbeae82b..ef6184b8a 100755 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/api/SchedulerJobApiConstants.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/api/SchedulerJobApiConstants.java @@ -6,7 +6,6 @@ import java.util.Set; public class SchedulerJobApiConstants { - public static final String JOB_RESOURCE_NAME = "schedulerjob"; // response parameters public static final String jobIdentifierParamName = "jobId"; @@ -16,7 +15,6 @@ public class SchedulerJobApiConstants { 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"; @@ -25,16 +23,27 @@ public class SchedulerJobApiConstants { public static final String jobRunErrorMessageParamName = "jobRunErrorMessage"; public static final String triggerTypeParamName = "triggerType"; public static final String jobRunErrorLogParamName = "jobRunErrorLog"; - public static final String cronExpressionParamName = "cronExpression"; + public static final String schedulerStatusParamName = "active"; public static final Set JOB_DETAIL_RESPONSE_DATA_PARAMETERS = new HashSet(Arrays.asList(jobIdentifierParamName, - displayNameParamName, nextRunTimeParamName, initializingErrorParamName, jobActiveStatusParamName,currentlyRunningParamName, lastRunHistoryObjParamName)); + displayNameParamName, nextRunTimeParamName, initializingErrorParamName, jobActiveStatusParamName, currentlyRunningParamName, + lastRunHistoryObjParamName)); public static final Set JOB_HISTORY_RESPONSE_DATA_PARAMETERS = new HashSet(Arrays.asList(versionParamName, jobRunStartTimeParamName, jobRunEndTimeParamName, statusParamName, jobRunErrorMessageParamName, triggerTypeParamName, jobRunErrorLogParamName)); - - public static final Set JOB_UPDATE_REQUEST_DATA_PARAMETERS = new HashSet(Arrays.asList(displayNameParamName,jobActiveStatusParamName,cronExpressionParamName)); + public static final Set JOB_UPDATE_REQUEST_DATA_PARAMETERS = new HashSet(Arrays.asList(displayNameParamName, + jobActiveStatusParamName, cronExpressionParamName)); + + public static final Set SCHEDULER_DETAIL_RESPONSE_DATA_PARAMETERS = new HashSet(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"; } diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/api/SchedulerJobApiResource.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/api/SchedulerJobApiResource.java index a71b2fe57..f978bdf89 100755 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/api/SchedulerJobApiResource.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/api/SchedulerJobApiResource.java @@ -15,11 +15,13 @@ 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.commands.domain.CommandWrapper; import org.mifosplatform.commands.service.CommandWrapperBuilder; import org.mifosplatform.commands.service.PortfolioCommandSourceWritePlatformService; import org.mifosplatform.infrastructure.core.api.ApiRequestParameterHelper; 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.ToApiJsonSerializer; import org.mifosplatform.infrastructure.core.service.Page; @@ -66,52 +68,50 @@ public class SchedulerJobApiResource { } @GET - @Path("{jobId}") + @Path("{" + SchedulerJobApiConstants.JOB_ID + "}") @Consumes({ 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); - 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); } @GET - @Path("{jobId}/runhistory") + @Path("{" + SchedulerJobApiConstants.JOB_ID + "}/" + SchedulerJobApiConstants.JOB_RUN_HISTORY) @Consumes({ 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("orderBy") final String orderBy, @QueryParam("sortOrder") final String sortOrder) { final SearchParameters searchParameters = SearchParameters.forPagination(offset, limit, orderBy, sortOrder); Page 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, SchedulerJobApiConstants.JOB_HISTORY_RESPONSE_DATA_PARAMETERS); } @POST - @Path("{jobId}/executeJob") + @Path("{" + SchedulerJobApiConstants.JOB_ID + "}") @Consumes({ MediaType.APPLICATION_JSON }) @Produces({ MediaType.APPLICATION_JSON }) - public Response executeJob(@PathParam("jobId") final Long jobId) { - /** - * 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 - **/ + public Response executeJob(@PathParam(SchedulerJobApiConstants.JOB_ID) final Long jobId, + @QueryParam(SchedulerJobApiConstants.COMMAND) final String commandParam) { Response response = Response.status(400).build(); - jobRegisterService.executeJob(jobId); - response = Response.status(202).build(); + if (is(commandParam, SchedulerJobApiConstants.COMMAND_EXECUTE_JOB)) { + this.jobRegisterService.executeJob(jobId); + response = Response.status(202).build(); + } else { + throw new UnrecognizedQueryParamException(SchedulerJobApiConstants.COMMAND, commandParam); + } return response; } @PUT - @Path("{jobId}") + @Path("{" + SchedulerJobApiConstants.JOB_ID + "}") @Consumes({ 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() // .updateJobDetail(jobId) // @@ -125,4 +125,9 @@ public class SchedulerJobApiResource { } return this.toApiJsonSerializer.serialize(result); } + + private boolean is(final String commandParam, final String commandValue) { + return StringUtils.isNotBlank(commandParam) && commandParam.trim().equalsIgnoreCase(commandValue); + } + } diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/data/SchedulerDetailData.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/data/SchedulerDetailData.java new file mode 100755 index 000000000..e2b339221 --- /dev/null +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/data/SchedulerDetailData.java @@ -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; + } +} diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterService.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterService.java index 44d0cf426..7006d8d5d 100755 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterService.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterService.java @@ -1,15 +1,18 @@ package org.mifosplatform.infrastructure.jobs.service; + public interface JobRegisterService { public void executeJob(Long jobId); public void rescheduleJob(Long jobId); - public void stopScheduler(); + public void pauseScheduler(); public void startScheduler(); public boolean isSchedulerRunning(); + public void stopScheduler(String name); + } diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterServiceImpl.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterServiceImpl.java index 0cf1b06a3..8ab9222fe 100644 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterServiceImpl.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/JobRegisterServiceImpl.java @@ -1,5 +1,6 @@ package org.mifosplatform.infrastructure.jobs.service; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -9,6 +10,7 @@ import java.util.TimeZone; import javax.annotation.PostConstruct; 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.jobs.annotation.CronMethodParser; import org.mifosplatform.infrastructure.jobs.domain.ScheduledJobDetail; @@ -76,28 +78,35 @@ public class JobRegisterServiceImpl implements JobRegisterService { 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()); + jobDataMap.put(SchedulerServiceConstants.TRIGGER_TYPE_REFERENCE, SchedulerServiceConstants.TRIGGER_TYPE_APPLICATION); + jobDataMap.put(SchedulerServiceConstants.TENANT_IDENTIFIER, ThreadLocalContextUtil.getTenant().getTenantIdentifier()); String key = scheduledJobDetail.getJobKey(); JobKey jobKey = constructJobKey(key); String schedulerName = getSchedulerName(scheduledJobDetail); Scheduler scheduler = schedulers.get(schedulerName); - if (scheduler != null && scheduler.checkExists(jobKey)) { - try { - scheduler.triggerJob(jobKey, jobDataMap); - } catch (SchedulerException e) { - logger.error(e.getMessage(), e); - } + if (scheduler == null || scheduler.isInStandbyMode() || !scheduler.checkExists(jobKey)) { + JobDetail jobDetail = createJobDetail(scheduledJobDetail); + String tempSchedulerName = "temp" + scheduledJobDetail.getId(); + List> listenerClasses = new ArrayList>(2); + 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 { - 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 { String jobIdentity = scheduledJobDetail.getJobKey(); JobKey jobKey = constructJobKey(jobIdentity); @@ -116,7 +125,7 @@ public class JobRegisterServiceImpl implements JobRegisterService { } @Override - public void stopScheduler() { + public void pauseScheduler() { schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), false); for (Map.Entry schedulerEntry : this.schedulers.entrySet()) { try { @@ -132,7 +141,6 @@ public class JobRegisterServiceImpl implements JobRegisterService { @Override public void startScheduler() { - schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), true); for (Map.Entry schedulerEntry : this.schedulers.entrySet()) { try { String schedulerName = schedulerEntry.getValue().getSchedulerName(); @@ -143,26 +151,20 @@ public class JobRegisterServiceImpl implements JobRegisterService { logger.error(e.getMessage(), e); } } + schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), true); } @Override - public void rescheduleJob(Long jobId) { + public void rescheduleJob(final Long jobId) { ScheduledJobDetail scheduledJobDetail = this.schedularWritePlatformService.findByJobId(jobId); rescheduleJob(scheduledJobDetail); } @Override - public void executeJob(Long jobId) { - try { - ScheduledJobDetail scheduledJobDetail = this.schedularWritePlatformService.findByJobId(jobId); - if (scheduledJobDetail != null) { - executeJob(scheduledJobDetail); - } else { - throw new JobNotFoundException(String.valueOf(jobId)); - } - } catch (JobNotFoundException e) { - throw new JobNotFoundException(String.valueOf(jobId)); - } + public void executeJob(final Long jobId) { + ScheduledJobDetail scheduledJobDetail = this.schedularWritePlatformService.findByJobId(jobId); + if (scheduledJobDetail == null) { throw new JobNotFoundException(String.valueOf(jobId)); } + executeJob(scheduledJobDetail); } @Override @@ -170,27 +172,27 @@ public class JobRegisterServiceImpl implements JobRegisterService { return this.schedulerStatus.get(ThreadLocalContextUtil.getTenant().getId()); } - private boolean isCurrentTenantsScheduler(String schedulerName) { - int beginIndex = SchedularServiceConstants.SCHEDULER.length(); + private boolean isCurrentTenantsScheduler(final String schedulerName) { + int beginIndex = SchedulerServiceConstants.SCHEDULER.length(); int endIndex = schedulerName.length(); - if (schedulerName.contains(SchedularServiceConstants.SCHEDULER_GROUP)) { - endIndex = schedulerName.indexOf(SchedularServiceConstants.SCHEDULER_GROUP); + if (schedulerName.contains(SchedulerServiceConstants.SCHEDULER_GROUP)) { + endIndex = schedulerName.indexOf(SchedulerServiceConstants.SCHEDULER_GROUP); } String tenantId = schedulerName.substring(beginIndex, endIndex); MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant(); return tenant.getId() == Long.valueOf(tenantId); } - private void scheduleJob(ScheduledJobDetail scheduledJobDetails) { + private void scheduleJob(final ScheduledJobDetail scheduledJobDetails) { if (!scheduledJobDetails.isActiveSchedular()) { scheduledJobDetails.updateNextRunTime(null); scheduledJobDetails.updateCurrentlyRunningStatus(false); return; } try { - Scheduler scheduler = getScheduler(scheduledJobDetails); JobDetail jobDetail = createJobDetail(scheduledJobDetails); Trigger trigger = createTrigger(scheduledJobDetails, jobDetail); + Scheduler scheduler = getScheduler(scheduledJobDetails); scheduler.scheduleJob(jobDetail, trigger); scheduledJobDetails.updateJobKey(getJobKeyAsString(jobDetail.getKey())); scheduledJobDetails.updateNextRunTime(trigger.getNextFireTime()); @@ -203,34 +205,49 @@ public class JobRegisterServiceImpl implements JobRegisterService { scheduledJobDetails.updateCurrentlyRunningStatus(false); } - private Scheduler getScheduler(ScheduledJobDetail scheduledJobDetail) throws Exception { + private Scheduler getScheduler(final ScheduledJobDetail scheduledJobDetail) throws Exception { String schedulername = getSchedulerName(scheduledJobDetail); Scheduler scheduler = this.schedulers.get(schedulername); if (scheduler == null) { - int noOfThreads = SchedularServiceConstants.DEFAULT_THREAD_COUNT; + int noOfThreads = SchedulerServiceConstants.DEFAULT_THREAD_COUNT; 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); } 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); MifosPlatformTenant tenant = ThreadLocalContextUtil.getTenant(); - sb.append(SchedularServiceConstants.SCHEDULER).append(tenant.getId()); + sb.append(SchedulerServiceConstants.SCHEDULER).append(tenant.getId()); if (scheduledJobDetail.getSchedulerGroup() > 0) { - sb.append(SchedularServiceConstants.SCHEDULER_GROUP).append(scheduledJobDetail.getSchedulerGroup()); + sb.append(SchedulerServiceConstants.SCHEDULER_GROUP).append(scheduledJobDetail.getSchedulerGroup()); } return sb.toString(); } - private Scheduler createScheduler(String name, int noOfThreads) throws Exception { + private Scheduler createScheduler(final String name, final int noOfThreads, List> listenerClasses) + throws Exception { SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean(); schedulerFactoryBean.setSchedulerName(name); - schedulerFactoryBean.setGlobalJobListeners(getGlobalListener()); + if (listenerClasses == null) { + listenerClasses = new ArrayList>(1); + listenerClasses.add(SchedulerJobListener.class); + } + schedulerFactoryBean.setGlobalJobListeners(getGlobalListener(listenerClasses)); Properties quartzProperties = new Properties(); quartzProperties.put(SchedulerFactoryBean.PROP_THREAD_COUNT, Integer.toString(noOfThreads)); schedulerFactoryBean.setQuartzProperties(quartzProperties); @@ -239,21 +256,25 @@ public class JobRegisterServiceImpl implements JobRegisterService { return schedulerFactoryBean.getScheduler(); } - private JobListener[] getGlobalListener() throws ClassNotFoundException { - JobListener listener = (JobListener) getBeanObject(SchedulerJobListener.class); - JobListener[] listenerArray = { listener }; - return listenerArray; + private JobListener[] getGlobalListener(List> listenerClasses) throws ClassNotFoundException { + List listeners = new ArrayList(listenerClasses.size()); + for (Class listenerClass : listenerClasses) { + 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(); - String[] jobDetails = CronMethodParser.findTargetMethodDetails(scheduledJobDetails.getJobName()); + String[] jobDetails = CronMethodParser.findTargetMethodDetails(scheduledJobDetail.getJobName()); Object targetObject = getBeanObject(Class.forName(jobDetails[CronMethodParser.CLASS_INDEX])); MethodInvokingJobDetailFactoryBean jobDetailFactoryBean = new MethodInvokingJobDetailFactoryBean(); - jobDetailFactoryBean.setName(scheduledJobDetails.getJobName() + "JobDetail" + tenant.getId()); + jobDetailFactoryBean.setName(scheduledJobDetail.getJobName() + "JobDetail" + tenant.getId()); jobDetailFactoryBean.setTargetObject(targetObject); jobDetailFactoryBean.setTargetMethod(jobDetails[CronMethodParser.METHOD_INDEX]); - jobDetailFactoryBean.setGroup(scheduledJobDetails.getGroupName()); + jobDetailFactoryBean.setGroup(scheduledJobDetail.getGroupName()); jobDetailFactoryBean.setConcurrent(false); jobDetailFactoryBean.afterPropertiesSet(); return jobDetailFactoryBean.getObject(); @@ -280,13 +301,13 @@ public class JobRegisterServiceImpl implements JobRegisterService { * @param scheduledJobDetails * @return */ - private Trigger createTrigger(ScheduledJobDetail scheduledJobDetails, JobDetail jobDetail) { + private Trigger createTrigger(final ScheduledJobDetail scheduledJobDetails, final 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()); + jobDataMap.put(SchedulerServiceConstants.TENANT_IDENTIFIER, tenant.getTenantIdentifier()); cronTriggerFactoryBean.setJobDataMap(jobDataMap); TimeZone timeZone = TimeZone.getTimeZone(tenant.getTimezoneId()); cronTriggerFactoryBean.setTimeZone(timeZone); @@ -311,12 +332,12 @@ public class JobRegisterServiceImpl implements JobRegisterService { return sb.toString(); } - private String getJobKeyAsString(JobKey jobKey) { - return jobKey.getName() + SchedularServiceConstants.JOB_KEY_SEPERATOR + jobKey.getGroup(); + private String getJobKeyAsString(final JobKey jobKey) { + return jobKey.getName() + SchedulerServiceConstants.JOB_KEY_SEPERATOR + jobKey.getGroup(); } - private JobKey constructJobKey(String Key) { - String[] keyParams = Key.split(SchedularServiceConstants.JOB_KEY_SEPERATOR); + private JobKey constructJobKey(final String Key) { + String[] keyParams = Key.split(SchedulerServiceConstants.JOB_KEY_SEPERATOR); JobKey JobKey = new JobKey(keyParams[0], keyParams[1]); return JobKey; } diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerJobListener.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerJobListener.java index 77e53aa07..a83811f07 100644 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerJobListener.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerJobListener.java @@ -27,7 +27,7 @@ public class SchedulerJobListener implements JobListener { private int stackTraceLevel = 0; - private final String name = SchedularServiceConstants.DEFAULT_LISTENER_NAME; + private final String name = SchedulerServiceConstants.DEFAULT_LISTENER_NAME; private final SchedularWritePlatformService schedularService; @@ -46,11 +46,11 @@ public class SchedulerJobListener implements JobListener { @Override 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); ThreadLocalContextUtil.setTenant(tenant); 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); if (scheduledJobDetail.isCurrentlyRunning()) { throw new JobInProcessExecution(scheduledJobDetail.getJobName()); } scheduledJobDetail.updateCurrentlyRunningStatus(true); @@ -67,14 +67,14 @@ public class SchedulerJobListener implements JobListener { public void jobWasExecuted(final JobExecutionContext context, final JobExecutionException jobException) { Trigger trigger = context.getTrigger(); 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); Long version = schedularService.fetchMaxVersionBy(jobKey) + 1; - String status = SchedularServiceConstants.STATUS_SUCCESS; + String status = SchedulerServiceConstants.STATUS_SUCCESS; String errorMessage = null; String errorLog = null; if (jobException != null) { - status = SchedularServiceConstants.STATUS_FAILED; + status = SchedulerServiceConstants.STATUS_FAILED; stackTraceLevel = 0; Throwable throwable = getCauseFromException(jobException); stackTraceLevel = 0; @@ -89,11 +89,11 @@ public class SchedulerJobListener implements JobListener { errorLog = sb.toString(); } - String triggerType = SchedularServiceConstants.TRIGGER_TYPE_CRON; - if (context.getMergedJobDataMap().containsKey(SchedularServiceConstants.TRIGGER_TYPE_REFERENCE)) { - triggerType = context.getMergedJobDataMap().getString(SchedularServiceConstants.TRIGGER_TYPE_REFERENCE); + String triggerType = SchedulerServiceConstants.TRIGGER_TYPE_CRON; + if (context.getMergedJobDataMap().containsKey(SchedulerServiceConstants.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()); } @@ -109,11 +109,11 @@ public class SchedulerJobListener implements JobListener { } private Throwable getCauseFromException(final Throwable exception) { - if (stackTraceLevel <= SchedularServiceConstants.STACK_TRACE_LEVEL + if (stackTraceLevel <= SchedulerServiceConstants.STACK_TRACE_LEVEL && exception.getCause() != null - && (exception.getCause().toString().contains(SchedularServiceConstants.SCHEDULAR_EXCEPTION) - || exception.getCause().toString().contains(SchedularServiceConstants.JOB_EXECUTION_EXCEPTION) || exception - .getCause().toString().contains(SchedularServiceConstants.JOB_METHOD_INVOCATION_FAILED_EXCEPTION))) { + && (exception.getCause().toString().contains(SchedulerServiceConstants.SCHEDULER_EXCEPTION) + || exception.getCause().toString().contains(SchedulerServiceConstants.JOB_EXECUTION_EXCEPTION) || exception + .getCause().toString().contains(SchedulerServiceConstants.JOB_METHOD_INVOCATION_FAILED_EXCEPTION))) { stackTraceLevel++; return getCauseFromException(exception.getCause()); } else if (exception.getCause() != null) { return exception.getCause(); } diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularServiceConstants.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerServiceConstants.java similarity index 86% rename from mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularServiceConstants.java rename to mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerServiceConstants.java index 46866c99f..f95c21ce8 100644 --- a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedularServiceConstants.java +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerServiceConstants.java @@ -1,12 +1,12 @@ package org.mifosplatform.infrastructure.jobs.service; -public interface SchedularServiceConstants { +public interface SchedulerServiceConstants { 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"; - 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_METHOD_INVOCATION_FAILED_EXCEPTION = "JobMethodInvocationFailedException"; public static final String STATUS_SUCCESS = "success"; @@ -18,5 +18,6 @@ public interface SchedularServiceConstants { public static final String SCHEDULER_GROUP = "group"; public static final int DEFAULT_THREAD_COUNT = 7; public static final int GROUP_THREAD_COUNT = 1; + public static final String SCHEDULER_NAME = "schedulerName"; } \ No newline at end of file diff --git a/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerStopListener.java b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerStopListener.java new file mode 100755 index 000000000..cb7ab74d1 --- /dev/null +++ b/mifosng-provider/src/main/java/org/mifosplatform/infrastructure/jobs/service/SchedulerStopListener.java @@ -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(); + } + } + +}