Merge pull request #486 from pramodn02/MIFOSX-506_7
added scheduler suspend API
This commit is contained in:
Vishwas Babu A J
Executable
+71
@@ -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);
|
||||
}
|
||||
}
|
||||
+15
-6
@@ -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<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,
|
||||
jobRunStartTimeParamName, jobRunEndTimeParamName, statusParamName, jobRunErrorMessageParamName, triggerTypeParamName,
|
||||
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";
|
||||
}
|
||||
|
||||
+24
-19
@@ -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<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,
|
||||
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);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Executable
+11
@@ -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;
|
||||
}
|
||||
}
|
||||
+4
-1
@@ -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);
|
||||
|
||||
}
|
||||
|
||||
+98
-58
@@ -1,5 +1,7 @@
|
||||
package org.mifosplatform.infrastructure.jobs.service;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
@@ -9,6 +11,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 +79,36 @@ 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<Class<? extends JobListener>> listenerClasses = new ArrayList<Class<? extends JobListener>>(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) {
|
||||
e.printStackTrace();
|
||||
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 +127,7 @@ public class JobRegisterServiceImpl implements JobRegisterService {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void stopScheduler() {
|
||||
public void pauseScheduler() {
|
||||
schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), false);
|
||||
for (Map.Entry<String, Scheduler> schedulerEntry : this.schedulers.entrySet()) {
|
||||
try {
|
||||
@@ -132,7 +143,6 @@ public class JobRegisterServiceImpl implements JobRegisterService {
|
||||
|
||||
@Override
|
||||
public void startScheduler() {
|
||||
schedulerStatus.put(ThreadLocalContextUtil.getTenant().getId(), true);
|
||||
for (Map.Entry<String, Scheduler> schedulerEntry : this.schedulers.entrySet()) {
|
||||
try {
|
||||
String schedulerName = schedulerEntry.getValue().getSchedulerName();
|
||||
@@ -143,26 +153,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 +174,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 +207,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<Class<? extends JobListener>> listenerClasses)
|
||||
throws Exception {
|
||||
SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();
|
||||
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();
|
||||
quartzProperties.put(SchedulerFactoryBean.PROP_THREAD_COUNT, Integer.toString(noOfThreads));
|
||||
schedulerFactoryBean.setQuartzProperties(quartzProperties);
|
||||
@@ -239,21 +258,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<Class<? extends JobListener>> listenerClasses) throws ClassNotFoundException {
|
||||
List<JobListener> listeners = new ArrayList<JobListener>(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();
|
||||
@@ -265,12 +288,29 @@ public class JobRegisterServiceImpl implements JobRegisterService {
|
||||
* @throws ClassNotFoundException
|
||||
*/
|
||||
private Object getBeanObject(Class<?> classType) throws ClassNotFoundException {
|
||||
String[] beanNames = this.applicationContext.getBeanNamesForType(classType);
|
||||
Object targetObject = this.applicationContext.getBean(beanNames[0]);
|
||||
List<Class<?>> typesList = new ArrayList<Class<?>>();
|
||||
Class<?>[] interfaceType = classType.getInterfaces();
|
||||
if (interfaceType.length > 0) {
|
||||
typesList.addAll(Arrays.asList(interfaceType));
|
||||
} else {
|
||||
Class<?> superclassType = classType;
|
||||
while (!Object.class.getName().equals(superclassType.getSuperclass().getName())) {
|
||||
superclassType = superclassType.getSuperclass();
|
||||
}
|
||||
typesList.add(superclassType);
|
||||
}
|
||||
List<String> beanNames = new ArrayList<String>();
|
||||
for (Class<?> clazz : typesList) {
|
||||
beanNames.addAll(Arrays.asList(this.applicationContext.getBeanNamesForType(clazz)));
|
||||
}
|
||||
Object targetObject = null;
|
||||
for (String beanName : beanNames) {
|
||||
Object nextObject = this.applicationContext.getBean(beanName);
|
||||
if (nextObject != null && !targetObject.getClass().isAssignableFrom(nextObject.getClass())) {
|
||||
String targetObjName = nextObject.toString();
|
||||
targetObjName = targetObjName.substring(0, targetObjName.lastIndexOf("@"));
|
||||
if (classType.getName().equals(targetObjName)) {
|
||||
targetObject = nextObject;
|
||||
break;
|
||||
}
|
||||
}
|
||||
return targetObject;
|
||||
@@ -280,13 +320,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 +351,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;
|
||||
}
|
||||
|
||||
+14
-14
@@ -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(); }
|
||||
|
||||
+3
-2
@@ -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";
|
||||
|
||||
}
|
||||
+55
@@ -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();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user