You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@syncope.apache.org by il...@apache.org on 2016/02/24 16:50:07 UTC
[2/4] syncope git commit: [SYNCOPE-744][SYNCOPE-750] Basic work done
for jobs and executions widget
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/logic/src/main/java/org/apache/syncope/core/logic/init/JobInstanceLoaderImpl.java
----------------------------------------------------------------------
diff --git a/core/logic/src/main/java/org/apache/syncope/core/logic/init/JobInstanceLoaderImpl.java b/core/logic/src/main/java/org/apache/syncope/core/logic/init/JobInstanceLoaderImpl.java
deleted file mode 100644
index 29bdba7..0000000
--- a/core/logic/src/main/java/org/apache/syncope/core/logic/init/JobInstanceLoaderImpl.java
+++ /dev/null
@@ -1,335 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.syncope.core.logic.init;
-
-import java.text.ParseException;
-import java.util.Collections;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.lang3.tuple.ImmutablePair;
-import org.apache.commons.lang3.tuple.Pair;
-import org.apache.syncope.common.lib.SyncopeConstants;
-import org.apache.syncope.common.lib.types.TaskType;
-import org.apache.syncope.core.logic.SystemLoadReporterJob;
-import org.apache.syncope.core.persistence.api.dao.ConfDAO;
-import org.apache.syncope.core.persistence.api.dao.NotFoundException;
-import org.apache.syncope.core.persistence.api.dao.ReportDAO;
-import org.apache.syncope.core.persistence.api.dao.TaskDAO;
-import org.apache.syncope.core.persistence.api.entity.Report;
-import org.apache.syncope.core.persistence.api.entity.conf.CPlainAttr;
-import org.apache.syncope.core.persistence.api.entity.task.PushTask;
-import org.apache.syncope.core.persistence.api.entity.task.SchedTask;
-import org.apache.syncope.core.persistence.api.entity.task.SyncTask;
-import org.apache.syncope.core.persistence.api.entity.task.Task;
-import org.apache.syncope.core.provisioning.api.job.JobInstanceLoader;
-import org.apache.syncope.core.provisioning.api.job.JobNamer;
-import org.apache.syncope.core.logic.notification.NotificationJob;
-import org.apache.syncope.core.logic.report.ReportJob;
-import org.apache.syncope.core.misc.security.AuthContextUtils;
-import org.apache.syncope.core.misc.spring.ApplicationContextProvider;
-import org.apache.syncope.core.persistence.api.SyncopeLoader;
-import org.apache.syncope.core.persistence.api.DomainsHolder;
-import org.apache.syncope.core.provisioning.java.job.TaskJob;
-import org.apache.syncope.core.provisioning.java.sync.PushJobDelegate;
-import org.apache.syncope.core.provisioning.java.sync.SyncJobDelegate;
-import org.quartz.CronScheduleBuilder;
-import org.quartz.Job;
-import org.quartz.JobBuilder;
-import org.quartz.JobDataMap;
-import org.quartz.JobExecutionContext;
-import org.quartz.JobKey;
-import org.quartz.Scheduler;
-import org.quartz.SchedulerException;
-import org.quartz.TriggerBuilder;
-import org.quartz.TriggerKey;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.BeanCreationException;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.support.AbstractBeanDefinition;
-import org.springframework.scheduling.quartz.SchedulerFactoryBean;
-import org.springframework.stereotype.Component;
-import org.springframework.transaction.annotation.Transactional;
-
-@Component
-public class JobInstanceLoaderImpl implements JobInstanceLoader, SyncopeLoader {
-
- private static final Logger LOG = LoggerFactory.getLogger(JobInstanceLoader.class);
-
- @Autowired
- private DomainsHolder domainsHolder;
-
- @Autowired
- private SchedulerFactoryBean scheduler;
-
- @Autowired
- private TaskDAO taskDAO;
-
- @Autowired
- private ReportDAO reportDAO;
-
- @Autowired
- private ConfDAO confDAO;
-
- private void registerJob(
- final String jobName, final Job jobInstance,
- final String cronExpression, final Date startAt,
- final Map<String, Object> jobMap)
- throws SchedulerException, ParseException {
-
- synchronized (scheduler.getScheduler()) {
- boolean jobAlreadyRunning = false;
- for (JobExecutionContext jobCtx : scheduler.getScheduler().getCurrentlyExecutingJobs()) {
- if (jobName.equals(jobCtx.getJobDetail().getKey().getName())
- && Scheduler.DEFAULT_GROUP.equals(jobCtx.getJobDetail().getKey().getGroup())) {
-
- jobAlreadyRunning = true;
-
- LOG.debug("Job {} already running, cancel", jobCtx.getJobDetail().getKey());
- }
- }
-
- if (jobAlreadyRunning) {
- return;
- }
- }
-
- // 0. unregister job
- unregisterJob(jobName);
-
- // 1. Job bean
- ApplicationContextProvider.getBeanFactory().registerSingleton(jobName, jobInstance);
-
- // 2. JobDetail bean
- JobBuilder jobDetailBuilder = JobBuilder.newJob(jobInstance.getClass()).
- withIdentity(jobName).
- usingJobData(new JobDataMap(jobMap));
-
- // 3. Trigger
- if (cronExpression == null && startAt == null) {
- // Jobs added with no trigger must be durable
- scheduler.getScheduler().addJob(jobDetailBuilder.storeDurably().build(), true);
- } else {
- TriggerBuilder<?> triggerBuilder;
-
- if (cronExpression == null) {
- triggerBuilder = TriggerBuilder.newTrigger().
- withIdentity(JobNamer.getTriggerName(jobName)).
- startAt(startAt);
- } else {
- triggerBuilder = TriggerBuilder.newTrigger().
- withIdentity(JobNamer.getTriggerName(jobName)).
- withSchedule(CronScheduleBuilder.cronSchedule(cronExpression));
-
- if (startAt == null) {
- triggerBuilder = triggerBuilder.startNow();
- } else {
- triggerBuilder = triggerBuilder.startAt(startAt);
- }
- }
-
- scheduler.getScheduler().scheduleJob(jobDetailBuilder.build(), triggerBuilder.build());
- }
- }
-
- @SuppressWarnings("unchecked")
- private <T> T createSpringBean(final Class<T> jobClass) {
- T jobInstance = null;
- for (int i = 0; i < 5 && jobInstance == null; i++) {
- LOG.debug("{} attempt to create Spring bean for {}", i, jobClass);
- try {
- jobInstance = (T) ApplicationContextProvider.getBeanFactory().
- createBean(jobClass, AbstractBeanDefinition.AUTOWIRE_BY_TYPE, false);
- LOG.debug("{} attempt to create Spring bean for {} succeeded", i, jobClass);
- } catch (BeanCreationException e) {
- LOG.error("Could not create Spring bean for {}", jobClass, e);
- try {
- Thread.sleep(1000);
- } catch (final InterruptedException ex) {
- // ignore
- }
- }
- }
- if (jobInstance == null) {
- throw new NotFoundException("Spring bean for " + jobClass);
- }
-
- return jobInstance;
- }
-
- @Override
- public Map<String, Object> registerJob(final SchedTask task, final Date startAt, final long interruptMaxRetries)
- throws SchedulerException, ParseException {
-
- TaskJob job = createSpringBean(TaskJob.class);
- job.setTaskKey(task.getKey());
-
- String jobDelegateClassName = task instanceof SyncTask
- ? SyncJobDelegate.class.getName()
- : task instanceof PushTask
- ? PushJobDelegate.class.getName()
- : task.getJobDelegateClassName();
-
- Map<String, Object> jobMap = new HashMap<>();
- jobMap.put(JobInstanceLoader.DOMAIN, AuthContextUtils.getDomain());
- jobMap.put(TaskJob.DELEGATE_CLASS_KEY, jobDelegateClassName);
- jobMap.put(TaskJob.INTERRUPT_MAX_RETRIES_KEY, interruptMaxRetries);
-
- registerJob(
- JobNamer.getJobName(task),
- job,
- task.getCronExpression(),
- startAt,
- jobMap);
- return jobMap;
- }
-
- @Override
- public void registerJob(final Report report, final Date startAt) throws SchedulerException, ParseException {
- ReportJob job = createSpringBean(ReportJob.class);
- job.setReportKey(report.getKey());
-
- Map<String, Object> jobMap = new HashMap<>();
- jobMap.put(JobInstanceLoader.DOMAIN, AuthContextUtils.getDomain());
-
- registerJob(JobNamer.getJobName(report), job, report.getCronExpression(), startAt, jobMap);
- }
-
- private void unregisterJob(final String jobName) {
- try {
- scheduler.getScheduler().unscheduleJob(new TriggerKey(jobName, Scheduler.DEFAULT_GROUP));
- scheduler.getScheduler().deleteJob(new JobKey(jobName, Scheduler.DEFAULT_GROUP));
- } catch (SchedulerException e) {
- LOG.error("Could not remove job " + jobName, e);
- }
-
- if (ApplicationContextProvider.getBeanFactory().containsSingleton(jobName)) {
- ApplicationContextProvider.getBeanFactory().destroySingleton(jobName);
- }
- }
-
- @Override
- public void unregisterJob(final Task task) {
- unregisterJob(JobNamer.getJobName(task));
- }
-
- @Override
- public void unregisterJob(final Report report) {
- unregisterJob(JobNamer.getJobName(report));
- }
-
- @Override
- public Integer getPriority() {
- return 200;
- }
-
- @Transactional
- @Override
- public void load() {
- final Pair<String, Long> notificationConf = AuthContextUtils.execWithAuthContext(SyncopeConstants.MASTER_DOMAIN,
- new AuthContextUtils.Executable<Pair<String, Long>>() {
-
- @Override
- public Pair<String, Long> exec() {
- String notificationJobCronExpression = StringUtils.EMPTY;
-
- CPlainAttr notificationJobCronExp =
- confDAO.find("notificationjob.cronExpression", NotificationJob.DEFAULT_CRON_EXP);
- if (!notificationJobCronExp.getValuesAsStrings().isEmpty()) {
- notificationJobCronExpression = notificationJobCronExp.getValuesAsStrings().get(0);
- }
-
- long interruptMaxRetries = confDAO.find("tasks.interruptMaxRetries", "1").getValues().get(0).
- getLongValue();
-
- return ImmutablePair.of(notificationJobCronExpression, interruptMaxRetries);
- }
- });
-
- for (String domain : domainsHolder.getDomains().keySet()) {
- AuthContextUtils.execWithAuthContext(domain, new AuthContextUtils.Executable<Void>() {
-
- @Override
- public Void exec() {
- // 1. jobs for SchedTasks
- Set<SchedTask> tasks = new HashSet<>(taskDAO.<SchedTask>findAll(TaskType.SCHEDULED));
- tasks.addAll(taskDAO.<SyncTask>findAll(TaskType.SYNCHRONIZATION));
- tasks.addAll(taskDAO.<PushTask>findAll(TaskType.PUSH));
- for (SchedTask task : tasks) {
- try {
- registerJob(task, task.getStartAt(), notificationConf.getRight());
- } catch (Exception e) {
- LOG.error("While loading job instance for task " + task.getKey(), e);
- }
- }
-
- // 2. ReportJobs
- for (Report report : reportDAO.findAll()) {
- try {
- registerJob(report, null);
- } catch (Exception e) {
- LOG.error("While loading job instance for report " + report.getName(), e);
- }
- }
-
- return null;
- }
- });
- }
-
- // 3. NotificationJob
- if (StringUtils.isBlank(notificationConf.getLeft())) {
- LOG.debug("Empty value provided for {}'s cron, not registering anything on Quartz",
- NotificationJob.class.getSimpleName());
- } else {
- LOG.debug("{}'s cron expression: {} - registering Quartz job and trigger",
- NotificationJob.class.getSimpleName(), notificationConf.getLeft());
-
- try {
- NotificationJob job = createSpringBean(NotificationJob.class);
- registerJob(
- "taskNotificationJob",
- job,
- notificationConf.getLeft(),
- null,
- Collections.<String, Object>emptyMap());
- } catch (Exception e) {
- LOG.error("While loading {} instance", NotificationJob.class.getSimpleName(), e);
- }
- }
-
- // 4. SystemLoadReporterJob (fixed schedule, every minute)
- LOG.debug("Registering {}", SystemLoadReporterJob.class);
- try {
- SystemLoadReporterJob job = createSpringBean(SystemLoadReporterJob.class);
- registerJob(
- "taskSystemLoadReporterJob",
- job,
- "0 * * * * ?",
- null,
- Collections.<String, Object>emptyMap());
- } catch (Exception e) {
- LOG.error("While loading {} instance", SystemLoadReporterJob.class.getSimpleName(), e);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/logic/src/main/java/org/apache/syncope/core/logic/init/JobManagerImpl.java
----------------------------------------------------------------------
diff --git a/core/logic/src/main/java/org/apache/syncope/core/logic/init/JobManagerImpl.java b/core/logic/src/main/java/org/apache/syncope/core/logic/init/JobManagerImpl.java
new file mode 100644
index 0000000..82134ed
--- /dev/null
+++ b/core/logic/src/main/java/org/apache/syncope/core/logic/init/JobManagerImpl.java
@@ -0,0 +1,387 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.syncope.core.logic.init;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import org.apache.commons.collections4.IterableUtils;
+import org.apache.commons.collections4.Predicate;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.syncope.common.lib.SyncopeConstants;
+import org.apache.syncope.common.lib.types.TaskType;
+import org.apache.syncope.core.logic.SystemLoadReporterJob;
+import org.apache.syncope.core.persistence.api.dao.ConfDAO;
+import org.apache.syncope.core.persistence.api.dao.NotFoundException;
+import org.apache.syncope.core.persistence.api.dao.ReportDAO;
+import org.apache.syncope.core.persistence.api.dao.TaskDAO;
+import org.apache.syncope.core.persistence.api.entity.Report;
+import org.apache.syncope.core.persistence.api.entity.conf.CPlainAttr;
+import org.apache.syncope.core.persistence.api.entity.task.PushTask;
+import org.apache.syncope.core.persistence.api.entity.task.SchedTask;
+import org.apache.syncope.core.persistence.api.entity.task.SyncTask;
+import org.apache.syncope.core.persistence.api.entity.task.Task;
+import org.apache.syncope.core.provisioning.api.job.JobNamer;
+import org.apache.syncope.core.logic.notification.NotificationJob;
+import org.apache.syncope.core.logic.report.ReportJob;
+import org.apache.syncope.core.misc.security.AuthContextUtils;
+import org.apache.syncope.core.misc.spring.ApplicationContextProvider;
+import org.apache.syncope.core.persistence.api.SyncopeLoader;
+import org.apache.syncope.core.persistence.api.DomainsHolder;
+import org.apache.syncope.core.provisioning.java.job.TaskJob;
+import org.apache.syncope.core.provisioning.java.sync.PushJobDelegate;
+import org.apache.syncope.core.provisioning.java.sync.SyncJobDelegate;
+import org.quartz.CronScheduleBuilder;
+import org.quartz.Job;
+import org.quartz.JobBuilder;
+import org.quartz.JobDataMap;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobKey;
+import org.quartz.Scheduler;
+import org.quartz.SchedulerException;
+import org.quartz.TriggerBuilder;
+import org.quartz.TriggerKey;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.BeanCreationException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.scheduling.quartz.SchedulerFactoryBean;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+import org.apache.syncope.core.provisioning.api.job.JobManager;
+import org.identityconnectors.common.IOUtil;
+import org.quartz.impl.jdbcjobstore.Constants;
+import org.springframework.jdbc.datasource.DataSourceUtils;
+
+@Component
+public class JobManagerImpl implements JobManager, SyncopeLoader {
+
+ private static final Logger LOG = LoggerFactory.getLogger(JobManager.class);
+
+ @Autowired
+ private DomainsHolder domainsHolder;
+
+ @Autowired
+ private SchedulerFactoryBean scheduler;
+
+ @Autowired
+ private TaskDAO taskDAO;
+
+ @Autowired
+ private ReportDAO reportDAO;
+
+ @Autowired
+ private ConfDAO confDAO;
+
+ private boolean isRunningHere(final JobKey jobKey) throws SchedulerException {
+ return IterableUtils.matchesAny(scheduler.getScheduler().getCurrentlyExecutingJobs(),
+ new Predicate<JobExecutionContext>() {
+
+ @Override
+ public boolean evaluate(final JobExecutionContext jec) {
+ return jobKey.equals(jec.getJobDetail().getKey());
+ }
+ });
+ }
+
+ private boolean isRunningElsewhere(final JobKey jobKey) throws SchedulerException {
+ if (!scheduler.getScheduler().getMetaData().isJobStoreClustered()) {
+ return false;
+ }
+
+ Connection conn = DataSourceUtils.getConnection(domainsHolder.getDomains().get(SyncopeConstants.MASTER_DOMAIN));
+ PreparedStatement stmt = null;
+ try {
+ stmt = conn.prepareStatement(
+ "SELECT 1 FROM " + Constants.DEFAULT_TABLE_PREFIX + "FIRED_TRIGGERS "
+ + "WHERE JOB_NAME = ? AND JOB_GROUP = ?");
+ stmt.setString(1, jobKey.getName());
+ stmt.setString(2, jobKey.getGroup());
+
+ return stmt.executeQuery().next();
+ } catch (SQLException e) {
+ throw new SchedulerException(e);
+ } finally {
+ IOUtil.quietClose(stmt);
+ IOUtil.quietClose(conn);
+ }
+ }
+
+ @Override
+ public boolean isRunning(final JobKey jobKey) throws SchedulerException {
+ return isRunningHere(jobKey) || isRunningElsewhere(jobKey);
+ }
+
+ private void registerJob(
+ final String jobName, final Job jobInstance,
+ final String cronExpression, final Date startAt,
+ final Map<String, Object> jobMap)
+ throws SchedulerException {
+
+ synchronized (scheduler.getScheduler()) {
+ boolean jobAlreadyRunning = false;
+ for (JobExecutionContext jobCtx : scheduler.getScheduler().getCurrentlyExecutingJobs()) {
+ if (jobName.equals(jobCtx.getJobDetail().getKey().getName())
+ && Scheduler.DEFAULT_GROUP.equals(jobCtx.getJobDetail().getKey().getGroup())) {
+
+ jobAlreadyRunning = true;
+
+ LOG.debug("Job {} already running, cancel", jobCtx.getJobDetail().getKey());
+ }
+ }
+
+ if (jobAlreadyRunning) {
+ return;
+ }
+ }
+
+ // 0. unregister job
+ unregisterJob(jobName);
+
+ // 1. Job bean
+ ApplicationContextProvider.getBeanFactory().registerSingleton(jobName, jobInstance);
+
+ // 2. JobDetail bean
+ JobBuilder jobDetailBuilder = JobBuilder.newJob(jobInstance.getClass()).
+ withIdentity(jobName).
+ usingJobData(new JobDataMap(jobMap));
+
+ // 3. Trigger
+ if (cronExpression == null && startAt == null) {
+ // Jobs added with no trigger must be durable
+ scheduler.getScheduler().addJob(jobDetailBuilder.storeDurably().build(), true);
+ } else {
+ TriggerBuilder<?> triggerBuilder;
+
+ if (cronExpression == null) {
+ triggerBuilder = TriggerBuilder.newTrigger().
+ withIdentity(JobNamer.getTriggerName(jobName)).
+ startAt(startAt);
+ } else {
+ triggerBuilder = TriggerBuilder.newTrigger().
+ withIdentity(JobNamer.getTriggerName(jobName)).
+ withSchedule(CronScheduleBuilder.cronSchedule(cronExpression));
+
+ if (startAt == null) {
+ triggerBuilder = triggerBuilder.startNow();
+ } else {
+ triggerBuilder = triggerBuilder.startAt(startAt);
+ }
+ }
+
+ scheduler.getScheduler().scheduleJob(jobDetailBuilder.build(), triggerBuilder.build());
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private <T> T createSpringBean(final Class<T> jobClass) {
+ T jobInstance = null;
+ for (int i = 0; i < 5 && jobInstance == null; i++) {
+ LOG.debug("{} attempt to create Spring bean for {}", i, jobClass);
+ try {
+ jobInstance = (T) ApplicationContextProvider.getBeanFactory().
+ createBean(jobClass, AbstractBeanDefinition.AUTOWIRE_BY_TYPE, false);
+ LOG.debug("{} attempt to create Spring bean for {} succeeded", i, jobClass);
+ } catch (BeanCreationException e) {
+ LOG.error("Could not create Spring bean for {}", jobClass, e);
+ try {
+ Thread.sleep(1000);
+ } catch (final InterruptedException ex) {
+ // ignore
+ }
+ }
+ }
+ if (jobInstance == null) {
+ throw new NotFoundException("Spring bean for " + jobClass);
+ }
+
+ return jobInstance;
+ }
+
+ @Override
+ public Map<String, Object> register(final SchedTask task, final Date startAt, final long interruptMaxRetries)
+ throws SchedulerException {
+
+ TaskJob job = createSpringBean(TaskJob.class);
+ job.setTaskKey(task.getKey());
+
+ String jobDelegateClassName = task instanceof SyncTask
+ ? SyncJobDelegate.class.getName()
+ : task instanceof PushTask
+ ? PushJobDelegate.class.getName()
+ : task.getJobDelegateClassName();
+
+ Map<String, Object> jobMap = new HashMap<>();
+ jobMap.put(JobManager.DOMAIN_KEY, AuthContextUtils.getDomain());
+ jobMap.put(TaskJob.DELEGATE_CLASS_KEY, jobDelegateClassName);
+ jobMap.put(INTERRUPT_MAX_RETRIES_KEY, interruptMaxRetries);
+
+ registerJob(
+ JobNamer.getJobKey(task).getName(),
+ job,
+ task.getCronExpression(),
+ startAt,
+ jobMap);
+ return jobMap;
+ }
+
+ @Override
+ public void register(final Report report, final Date startAt, final long interruptMaxRetries)
+ throws SchedulerException {
+
+ ReportJob job = createSpringBean(ReportJob.class);
+ job.setReportKey(report.getKey());
+
+ Map<String, Object> jobMap = new HashMap<>();
+ jobMap.put(JobManager.DOMAIN_KEY, AuthContextUtils.getDomain());
+ jobMap.put(INTERRUPT_MAX_RETRIES_KEY, interruptMaxRetries);
+
+ registerJob(JobNamer.getJobKey(report).getName(), job, report.getCronExpression(), startAt, jobMap);
+ }
+
+ private void unregisterJob(final String jobName) {
+ try {
+ scheduler.getScheduler().unscheduleJob(new TriggerKey(jobName, Scheduler.DEFAULT_GROUP));
+ scheduler.getScheduler().deleteJob(new JobKey(jobName, Scheduler.DEFAULT_GROUP));
+ } catch (SchedulerException e) {
+ LOG.error("Could not remove job " + jobName, e);
+ }
+
+ if (ApplicationContextProvider.getBeanFactory().containsSingleton(jobName)) {
+ ApplicationContextProvider.getBeanFactory().destroySingleton(jobName);
+ }
+ }
+
+ @Override
+ public void unregister(final Task task) {
+ unregisterJob(JobNamer.getJobKey(task).getName());
+ }
+
+ @Override
+ public void unregister(final Report report) {
+ unregisterJob(JobNamer.getJobKey(report).getName());
+ }
+
+ @Override
+ public Integer getPriority() {
+ return 200;
+ }
+
+ @Transactional
+ @Override
+ public void load() {
+ final Pair<String, Long> conf = AuthContextUtils.execWithAuthContext(
+ SyncopeConstants.MASTER_DOMAIN, new AuthContextUtils.Executable<Pair<String, Long>>() {
+
+ @Override
+ public Pair<String, Long> exec() {
+ String notificationJobCronExpression = StringUtils.EMPTY;
+
+ CPlainAttr notificationJobCronExp =
+ confDAO.find("notificationjob.cronExpression", NotificationJob.DEFAULT_CRON_EXP);
+ if (!notificationJobCronExp.getValuesAsStrings().isEmpty()) {
+ notificationJobCronExpression = notificationJobCronExp.getValuesAsStrings().get(0);
+ }
+
+ long interruptMaxRetries =
+ confDAO.find("tasks.interruptMaxRetries", "1").getValues().get(0).getLongValue();
+
+ return ImmutablePair.of(notificationJobCronExpression, interruptMaxRetries);
+ }
+ });
+
+ for (String domain : domainsHolder.getDomains().keySet()) {
+ AuthContextUtils.execWithAuthContext(domain, new AuthContextUtils.Executable<Void>() {
+
+ @Override
+ public Void exec() {
+ // 1. jobs for SchedTasks
+ Set<SchedTask> tasks = new HashSet<>(taskDAO.<SchedTask>findAll(TaskType.SCHEDULED));
+ tasks.addAll(taskDAO.<SyncTask>findAll(TaskType.SYNCHRONIZATION));
+ tasks.addAll(taskDAO.<PushTask>findAll(TaskType.PUSH));
+ for (SchedTask task : tasks) {
+ try {
+ register(task, task.getStartAt(), conf.getRight());
+ } catch (Exception e) {
+ LOG.error("While loading job instance for task " + task.getKey(), e);
+ }
+ }
+
+ // 2. jobs for Reports
+ for (Report report : reportDAO.findAll()) {
+ try {
+ register(report, null, conf.getRight());
+ } catch (Exception e) {
+ LOG.error("While loading job instance for report " + report.getName(), e);
+ }
+ }
+
+ return null;
+ }
+ });
+ }
+
+ Map<String, Object> jobMap = new HashMap<>();
+ jobMap.put(JobManager.DOMAIN_KEY, AuthContextUtils.getDomain());
+ jobMap.put(INTERRUPT_MAX_RETRIES_KEY, conf.getRight());
+
+ // 3. NotificationJob
+ if (StringUtils.isBlank(conf.getLeft())) {
+ LOG.debug("Empty value provided for {}'s cron, not registering anything on Quartz",
+ NotificationJob.class.getSimpleName());
+ } else {
+ LOG.debug("{}'s cron expression: {} - registering Quartz job and trigger",
+ NotificationJob.class.getSimpleName(), conf.getLeft());
+
+ try {
+ NotificationJob job = createSpringBean(NotificationJob.class);
+ registerJob(
+ NOTIFICATION_JOB.getName(),
+ job,
+ conf.getLeft(),
+ null,
+ jobMap);
+ } catch (Exception e) {
+ LOG.error("While loading {} instance", NotificationJob.class.getSimpleName(), e);
+ }
+ }
+
+ // 4. SystemLoadReporterJob (fixed schedule, every minute)
+ LOG.debug("Registering {}", SystemLoadReporterJob.class);
+ try {
+ SystemLoadReporterJob job = createSpringBean(SystemLoadReporterJob.class);
+ registerJob(
+ "systemLoadReporterJob",
+ job,
+ "0 * * * * ?",
+ null,
+ jobMap);
+ } catch (Exception e) {
+ LOG.error("While loading {} instance", SystemLoadReporterJob.class.getSimpleName(), e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/logic/src/main/java/org/apache/syncope/core/logic/notification/NotificationJob.java
----------------------------------------------------------------------
diff --git a/core/logic/src/main/java/org/apache/syncope/core/logic/notification/NotificationJob.java b/core/logic/src/main/java/org/apache/syncope/core/logic/notification/NotificationJob.java
index adba78d..7fea774 100644
--- a/core/logic/src/main/java/org/apache/syncope/core/logic/notification/NotificationJob.java
+++ b/core/logic/src/main/java/org/apache/syncope/core/logic/notification/NotificationJob.java
@@ -20,8 +20,7 @@ package org.apache.syncope.core.logic.notification;
import org.apache.syncope.core.misc.security.AuthContextUtils;
import org.apache.syncope.core.persistence.api.DomainsHolder;
-import org.quartz.DisallowConcurrentExecution;
-import org.quartz.Job;
+import org.apache.syncope.core.provisioning.java.job.AbstractInterruptableJob;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
@@ -35,8 +34,7 @@ import org.springframework.stereotype.Component;
* @see org.apache.syncope.core.persistence.api.entity.task.NotificationTask
*/
@Component
-@DisallowConcurrentExecution
-public class NotificationJob implements Job {
+public class NotificationJob extends AbstractInterruptableJob {
public enum Status {
@@ -57,6 +55,8 @@ public class NotificationJob implements Job {
@Override
public void execute(final JobExecutionContext context) throws JobExecutionException {
+ super.execute(context);
+
LOG.debug("Waking up...");
for (String domain : domainsHolder.getDomains().keySet()) {
@@ -81,5 +81,4 @@ public class NotificationJob implements Job {
LOG.debug("Sleeping again...");
}
-
}
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/logic/src/main/java/org/apache/syncope/core/logic/report/ReportJob.java
----------------------------------------------------------------------
diff --git a/core/logic/src/main/java/org/apache/syncope/core/logic/report/ReportJob.java b/core/logic/src/main/java/org/apache/syncope/core/logic/report/ReportJob.java
index 686953c..5786cfb 100644
--- a/core/logic/src/main/java/org/apache/syncope/core/logic/report/ReportJob.java
+++ b/core/logic/src/main/java/org/apache/syncope/core/logic/report/ReportJob.java
@@ -19,18 +19,16 @@
package org.apache.syncope.core.logic.report;
import org.apache.syncope.core.misc.security.AuthContextUtils;
-import org.apache.syncope.core.provisioning.api.job.JobInstanceLoader;
-import org.quartz.DisallowConcurrentExecution;
-import org.quartz.Job;
+import org.apache.syncope.core.provisioning.java.job.AbstractInterruptableJob;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.beans.factory.annotation.Autowired;
+import org.apache.syncope.core.provisioning.api.job.JobManager;
/**
* Quartz job for executing a given report.
*/
-@DisallowConcurrentExecution
-public class ReportJob implements Job {
+public class ReportJob extends AbstractInterruptableJob {
/**
* Key, set by the caller, for identifying the report to be executed.
@@ -51,23 +49,26 @@ public class ReportJob implements Job {
@Override
public void execute(final JobExecutionContext context) throws JobExecutionException {
+ super.execute(context);
+
try {
- AuthContextUtils.execWithAuthContext(context.getMergedJobDataMap().getString(JobInstanceLoader.DOMAIN),
+ AuthContextUtils.execWithAuthContext(context.getMergedJobDataMap().getString(JobManager.DOMAIN_KEY),
new AuthContextUtils.Executable<Void>() {
- @Override
- public Void exec() {
- try {
- delegate.execute(reportKey);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
+ @Override
+ public Void exec() {
+ try {
+ delegate.execute(reportKey);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
- return null;
- }
- });
+ return null;
+ }
+ });
} catch (RuntimeException e) {
throw new JobExecutionException(e.getCause());
}
}
+
}
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/persistence-api/src/main/java/org/apache/syncope/core/persistence/api/dao/ReportExecDAO.java
----------------------------------------------------------------------
diff --git a/core/persistence-api/src/main/java/org/apache/syncope/core/persistence/api/dao/ReportExecDAO.java b/core/persistence-api/src/main/java/org/apache/syncope/core/persistence/api/dao/ReportExecDAO.java
index c6b5318..dd79dd9 100644
--- a/core/persistence-api/src/main/java/org/apache/syncope/core/persistence/api/dao/ReportExecDAO.java
+++ b/core/persistence-api/src/main/java/org/apache/syncope/core/persistence/api/dao/ReportExecDAO.java
@@ -27,6 +27,8 @@ public interface ReportExecDAO extends DAO<ReportExec, Long> {
ReportExec find(Long key);
+ List<ReportExec> findRecent(int max);
+
ReportExec findLatestStarted(Report report);
ReportExec findLatestEnded(Report report);
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/persistence-api/src/main/java/org/apache/syncope/core/persistence/api/dao/TaskExecDAO.java
----------------------------------------------------------------------
diff --git a/core/persistence-api/src/main/java/org/apache/syncope/core/persistence/api/dao/TaskExecDAO.java b/core/persistence-api/src/main/java/org/apache/syncope/core/persistence/api/dao/TaskExecDAO.java
index faa8a14..4af1453 100644
--- a/core/persistence-api/src/main/java/org/apache/syncope/core/persistence/api/dao/TaskExecDAO.java
+++ b/core/persistence-api/src/main/java/org/apache/syncope/core/persistence/api/dao/TaskExecDAO.java
@@ -28,6 +28,8 @@ public interface TaskExecDAO extends DAO<TaskExec, Long> {
TaskExec find(Long key);
+ List<TaskExec> findRecent(int max);
+
<T extends Task> TaskExec findLatestStarted(T task);
<T extends Task> TaskExec findLatestEnded(T task);
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/persistence-jpa/src/main/java/org/apache/syncope/core/persistence/jpa/dao/JPAReportExecDAO.java
----------------------------------------------------------------------
diff --git a/core/persistence-jpa/src/main/java/org/apache/syncope/core/persistence/jpa/dao/JPAReportExecDAO.java b/core/persistence-jpa/src/main/java/org/apache/syncope/core/persistence/jpa/dao/JPAReportExecDAO.java
index 400f8a2..1e0af21 100644
--- a/core/persistence-jpa/src/main/java/org/apache/syncope/core/persistence/jpa/dao/JPAReportExecDAO.java
+++ b/core/persistence-jpa/src/main/java/org/apache/syncope/core/persistence/jpa/dao/JPAReportExecDAO.java
@@ -36,6 +36,16 @@ public class JPAReportExecDAO extends AbstractDAO<ReportExec, Long> implements R
return entityManager().find(JPAReportExec.class, key);
}
+ @Override
+ public List<ReportExec> findRecent(final int max) {
+ TypedQuery<ReportExec> query = entityManager().createQuery(
+ "SELECT e FROM " + JPAReportExec.class.getSimpleName() + " e "
+ + "WHERE e.end IS NOT NULL ORDER BY e.end DESC", ReportExec.class);
+ query.setMaxResults(max);
+
+ return query.getResultList();
+ }
+
private ReportExec findLatest(final Report report, final String field) {
TypedQuery<ReportExec> query = entityManager().createQuery(
"SELECT e FROM " + JPAReportExec.class.getSimpleName() + " e "
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/persistence-jpa/src/main/java/org/apache/syncope/core/persistence/jpa/dao/JPATaskExecDAO.java
----------------------------------------------------------------------
diff --git a/core/persistence-jpa/src/main/java/org/apache/syncope/core/persistence/jpa/dao/JPATaskExecDAO.java b/core/persistence-jpa/src/main/java/org/apache/syncope/core/persistence/jpa/dao/JPATaskExecDAO.java
index b44d4c7..c0fcacb 100644
--- a/core/persistence-jpa/src/main/java/org/apache/syncope/core/persistence/jpa/dao/JPATaskExecDAO.java
+++ b/core/persistence-jpa/src/main/java/org/apache/syncope/core/persistence/jpa/dao/JPATaskExecDAO.java
@@ -44,11 +44,20 @@ public class JPATaskExecDAO extends AbstractDAO<TaskExec, Long> implements TaskE
return entityManager().find(JPATaskExec.class, key);
}
+ @Override
+ public List<TaskExec> findRecent(final int max) {
+ TypedQuery<TaskExec> query = entityManager().createQuery(
+ "SELECT e FROM " + JPATaskExec.class.getSimpleName() + " e "
+ + "WHERE e.end IS NOT NULL ORDER BY e.end DESC", TaskExec.class);
+ query.setMaxResults(max);
+
+ return query.getResultList();
+ }
+
private <T extends Task> TaskExec findLatest(final T task, final String field) {
TypedQuery<TaskExec> query = entityManager().createQuery(
"SELECT e FROM " + JPATaskExec.class.getSimpleName() + " e "
- + "WHERE e.task=:task "
- + "ORDER BY e." + field + " DESC", TaskExec.class);
+ + "WHERE e.task=:task ORDER BY e." + field + " DESC", TaskExec.class);
query.setParameter("task", task);
query.setMaxResults(1);
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/provisioning-api/src/main/java/org/apache/syncope/core/provisioning/api/job/JobInstanceLoader.java
----------------------------------------------------------------------
diff --git a/core/provisioning-api/src/main/java/org/apache/syncope/core/provisioning/api/job/JobInstanceLoader.java b/core/provisioning-api/src/main/java/org/apache/syncope/core/provisioning/api/job/JobInstanceLoader.java
deleted file mode 100644
index 434d735..0000000
--- a/core/provisioning-api/src/main/java/org/apache/syncope/core/provisioning/api/job/JobInstanceLoader.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.syncope.core.provisioning.api.job;
-
-import java.text.ParseException;
-import java.util.Date;
-import java.util.Map;
-import org.apache.syncope.core.persistence.api.entity.Report;
-import org.apache.syncope.core.persistence.api.entity.task.SchedTask;
-import org.apache.syncope.core.persistence.api.entity.task.Task;
-import org.quartz.SchedulerException;
-
-public interface JobInstanceLoader {
-
- String DOMAIN = "domain";
-
- Map<String, Object> registerJob(final SchedTask task, final Date startAt, final long interruptMaxRetries)
- throws SchedulerException, ParseException;
-
- void registerJob(final Report report, final Date startAt)
- throws SchedulerException, ParseException;
-
- void unregisterJob(Task task);
-
- void unregisterJob(Report report);
-
-}
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/provisioning-api/src/main/java/org/apache/syncope/core/provisioning/api/job/JobManager.java
----------------------------------------------------------------------
diff --git a/core/provisioning-api/src/main/java/org/apache/syncope/core/provisioning/api/job/JobManager.java b/core/provisioning-api/src/main/java/org/apache/syncope/core/provisioning/api/job/JobManager.java
new file mode 100644
index 0000000..0e327fe
--- /dev/null
+++ b/core/provisioning-api/src/main/java/org/apache/syncope/core/provisioning/api/job/JobManager.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.syncope.core.provisioning.api.job;
+
+import java.util.Date;
+import java.util.Map;
+import org.apache.syncope.core.persistence.api.entity.Report;
+import org.apache.syncope.core.persistence.api.entity.task.SchedTask;
+import org.apache.syncope.core.persistence.api.entity.task.Task;
+import org.quartz.JobKey;
+import org.quartz.Scheduler;
+import org.quartz.SchedulerException;
+
+public interface JobManager {
+
+ String DOMAIN_KEY = "domain";
+
+ String INTERRUPT_MAX_RETRIES_KEY = "interruptMaxRetries";
+
+ JobKey NOTIFICATION_JOB = new JobKey("notificationJob", Scheduler.DEFAULT_GROUP);
+
+ boolean isRunning(JobKey jobKey) throws SchedulerException;
+
+ Map<String, Object> register(SchedTask task, Date startAt, long interruptMaxRetries)
+ throws SchedulerException;
+
+ void register(Report report, Date startAt, long interruptMaxRetries)
+ throws SchedulerException;
+
+ void unregister(Task task);
+
+ void unregister(Report report);
+
+}
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/provisioning-api/src/main/java/org/apache/syncope/core/provisioning/api/job/JobNamer.java
----------------------------------------------------------------------
diff --git a/core/provisioning-api/src/main/java/org/apache/syncope/core/provisioning/api/job/JobNamer.java b/core/provisioning-api/src/main/java/org/apache/syncope/core/provisioning/api/job/JobNamer.java
index 22d7b54..cc3cf00 100644
--- a/core/provisioning-api/src/main/java/org/apache/syncope/core/provisioning/api/job/JobNamer.java
+++ b/core/provisioning-api/src/main/java/org/apache/syncope/core/provisioning/api/job/JobNamer.java
@@ -22,6 +22,8 @@ import java.util.regex.Matcher;
import java.util.regex.Pattern;
import org.apache.syncope.core.persistence.api.entity.Report;
import org.apache.syncope.core.persistence.api.entity.task.Task;
+import org.quartz.JobKey;
+import org.quartz.Scheduler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -52,12 +54,12 @@ public final class JobNamer {
return getKeyFromJobName(name, "reportJob[0-9]+", 9);
}
- public static String getJobName(final Task task) {
- return "taskJob" + task.getKey();
+ public static JobKey getJobKey(final Task task) {
+ return new JobKey("taskJob" + task.getKey(), Scheduler.DEFAULT_GROUP);
}
- public static String getJobName(final Report report) {
- return "reportJob" + report.getKey();
+ public static JobKey getJobKey(final Report report) {
+ return new JobKey("reportJob" + report.getKey(), Scheduler.DEFAULT_GROUP);
}
public static String getTriggerName(final String jobName) {
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/data/ReportDataBinderImpl.java
----------------------------------------------------------------------
diff --git a/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/data/ReportDataBinderImpl.java b/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/data/ReportDataBinderImpl.java
index 0207d49..6057f0e 100644
--- a/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/data/ReportDataBinderImpl.java
+++ b/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/data/ReportDataBinderImpl.java
@@ -107,7 +107,7 @@ public class ReportDataBinderImpl implements ReportDataBinder {
reportTO.getExecutions().add(getReportExecTO(reportExec));
}
- String triggerName = JobNamer.getTriggerName(JobNamer.getJobName(report));
+ String triggerName = JobNamer.getTriggerName(JobNamer.getJobKey(report).getName());
try {
Trigger trigger = scheduler.getScheduler().getTrigger(new TriggerKey(triggerName, Scheduler.DEFAULT_GROUP));
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/data/TaskDataBinderImpl.java
----------------------------------------------------------------------
diff --git a/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/data/TaskDataBinderImpl.java b/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/data/TaskDataBinderImpl.java
index b07fffb..299b457 100644
--- a/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/data/TaskDataBinderImpl.java
+++ b/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/data/TaskDataBinderImpl.java
@@ -70,6 +70,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;
import org.springframework.stereotype.Component;
import org.apache.syncope.core.persistence.api.entity.task.PushTaskAnyFilter;
+import org.apache.syncope.core.persistence.api.entity.task.TaskUtilsFactory;
@Component
public class TaskDataBinderImpl implements TaskDataBinder {
@@ -102,6 +103,9 @@ public class TaskDataBinderImpl implements TaskDataBinder {
@Autowired
private SchedulerFactoryBean scheduler;
+ @Autowired
+ private TaskUtilsFactory taskUtilsFactory;
+
private void fill(final ProvisioningTask task, final AbstractProvisioningTaskTO taskTO) {
if (task instanceof PushTask && taskTO instanceof PushTaskTO) {
PushTask pushTask = (PushTask) task;
@@ -254,6 +258,7 @@ public class TaskDataBinderImpl implements TaskDataBinder {
if (execution.getTask() != null && execution.getTask().getKey() != null) {
executionTO.setTask(execution.getTask().getKey());
+ executionTO.setType(taskUtilsFactory.getInstance(execution.getTask()).getType());
}
return executionTO;
@@ -262,7 +267,7 @@ public class TaskDataBinderImpl implements TaskDataBinder {
private void setExecTime(final SchedTaskTO taskTO, final Task task) {
taskTO.setLastExec(taskTO.getStart());
- String triggerName = JobNamer.getTriggerName(JobNamer.getJobName(task));
+ String triggerName = JobNamer.getTriggerName(JobNamer.getJobKey(task).getName());
try {
Trigger trigger = scheduler.getScheduler().getTrigger(new TriggerKey(triggerName, Scheduler.DEFAULT_GROUP));
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/job/AbstractInterruptableJob.java
----------------------------------------------------------------------
diff --git a/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/job/AbstractInterruptableJob.java b/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/job/AbstractInterruptableJob.java
new file mode 100644
index 0000000..bd550b5
--- /dev/null
+++ b/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/job/AbstractInterruptableJob.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.syncope.core.provisioning.java.job;
+
+
+import java.util.Date;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.syncope.core.misc.utils.FormatUtils;
+import org.apache.syncope.core.provisioning.api.job.JobManager;
+import org.quartz.DisallowConcurrentExecution;
+import org.quartz.InterruptableJob;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+import org.quartz.UnableToInterruptJobException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@DisallowConcurrentExecution
+public abstract class AbstractInterruptableJob implements InterruptableJob {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractInterruptableJob.class);
+
+ /**
+ * The current running thread containing the task to be executed.
+ */
+ private final AtomicReference<Thread> runningThread = new AtomicReference<>();
+
+ private long interruptMaxRetries = 1;
+
+ @Override
+ public void execute(final JobExecutionContext context) throws JobExecutionException {
+ this.runningThread.set(Thread.currentThread());
+ this.interruptMaxRetries = context.getMergedJobDataMap().getLong(JobManager.INTERRUPT_MAX_RETRIES_KEY);
+ }
+
+ @Override
+ public void interrupt() throws UnableToInterruptJobException {
+ Thread thread = this.runningThread.getAndSet(null);
+ if (thread == null) {
+ LOG.warn("Unable to retrieve the thread of the current job execution");
+ } else {
+ LOG.info("Interrupting job from thread {} at {} ", thread.getId(), FormatUtils.format(new Date()));
+
+ if (interruptMaxRetries < 1) {
+ interruptMaxRetries = 1;
+ }
+ for (int i = 0; i < interruptMaxRetries && thread.isAlive(); i++) {
+ thread.interrupt();
+ }
+ // if the thread is still alive, it should be available in the next stop
+ if (thread.isAlive()) {
+ this.runningThread.set(thread);
+ }
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/job/TaskJob.java
----------------------------------------------------------------------
diff --git a/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/job/TaskJob.java b/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/job/TaskJob.java
index e3682de..4692e88 100644
--- a/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/job/TaskJob.java
+++ b/core/provisioning-java/src/main/java/org/apache/syncope/core/provisioning/java/job/TaskJob.java
@@ -18,34 +18,21 @@
*/
package org.apache.syncope.core.provisioning.java.job;
-import java.util.Date;
-import java.util.concurrent.atomic.AtomicReference;
import org.apache.commons.lang3.ClassUtils;
-import org.apache.syncope.core.misc.utils.FormatUtils;
import org.apache.syncope.core.misc.security.AuthContextUtils;
import org.apache.syncope.core.misc.spring.ApplicationContextProvider;
-import org.apache.syncope.core.provisioning.api.job.JobInstanceLoader;
import org.apache.syncope.core.provisioning.api.job.SchedTaskJobDelegate;
-import org.quartz.DisallowConcurrentExecution;
-import org.quartz.InterruptableJob;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
-import org.quartz.UnableToInterruptJobException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.apache.syncope.core.provisioning.api.job.JobManager;
-@DisallowConcurrentExecution
-public class TaskJob implements InterruptableJob {
-
- private static final Logger LOG = LoggerFactory.getLogger(TaskJob.class);
+public class TaskJob extends AbstractInterruptableJob {
public static final String DRY_RUN_JOBDETAIL_KEY = "dryRun";
public static final String DELEGATE_CLASS_KEY = "delegateClass";
- public static final String INTERRUPT_MAX_RETRIES_KEY = "interruptMaxRetries";
-
/**
* Task execution status.
*/
@@ -57,17 +44,10 @@ public class TaskJob implements InterruptableJob {
}
/**
- * The current running thread containing the task to be executed.
- */
- private final AtomicReference<Thread> runningThread = new AtomicReference<>();
-
- /**
* Key, set by the caller, for identifying the task to be executed.
*/
private Long taskKey;
- private long interruptMaxRetries = 1;
-
/**
* Task key setter.
*
@@ -79,53 +59,31 @@ public class TaskJob implements InterruptableJob {
@Override
public void execute(final JobExecutionContext context) throws JobExecutionException {
- this.runningThread.set(Thread.currentThread());
- this.interruptMaxRetries = context.getMergedJobDataMap().getLong(INTERRUPT_MAX_RETRIES_KEY);
+ super.execute(context);
try {
- AuthContextUtils.execWithAuthContext(context.getMergedJobDataMap().getString(JobInstanceLoader.DOMAIN),
+ AuthContextUtils.execWithAuthContext(context.getMergedJobDataMap().getString(JobManager.DOMAIN_KEY),
new AuthContextUtils.Executable<Void>() {
- @Override
- public Void exec() {
- try {
- Class<?> delegateClass =
+ @Override
+ public Void exec() {
+ try {
+ Class<?> delegateClass =
ClassUtils.getClass(context.getMergedJobDataMap().getString(DELEGATE_CLASS_KEY));
- ((SchedTaskJobDelegate) ApplicationContextProvider.getBeanFactory().
+ ((SchedTaskJobDelegate) ApplicationContextProvider.getBeanFactory().
createBean(delegateClass, AbstractBeanDefinition.AUTOWIRE_BY_NAME, false)).
execute(taskKey, context.getMergedJobDataMap().getBoolean(DRY_RUN_JOBDETAIL_KEY));
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
-
- return null;
- }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
}
+
+ return null;
+ }
+ }
);
} catch (RuntimeException e) {
throw new JobExecutionException(e.getCause());
}
}
-
- @Override
- public void interrupt() throws UnableToInterruptJobException {
- Thread thread = this.runningThread.getAndSet(null);
- if (thread == null) {
- LOG.warn("Unable to retrieve the thread of the current job execution");
- } else {
- LOG.info("Interrupting job from thread {} at {} ", thread.getId(), FormatUtils.format(new Date()));
-
- if (interruptMaxRetries < 1) {
- interruptMaxRetries = 1;
- }
- for (int i = 0; i < interruptMaxRetries && thread.isAlive(); i++) {
- thread.interrupt();
- }
- // if the thread is still alive, it should be available in the next stop
- if (thread.isAlive()) {
- this.runningThread.set(thread);
- }
- }
- }
}
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/NotificationServiceImpl.java
----------------------------------------------------------------------
diff --git a/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/NotificationServiceImpl.java b/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/NotificationServiceImpl.java
index dbd3461..43e6b5b 100644
--- a/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/NotificationServiceImpl.java
+++ b/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/NotificationServiceImpl.java
@@ -21,7 +21,9 @@ package org.apache.syncope.core.rest.cxf.service;
import java.net.URI;
import java.util.List;
import javax.ws.rs.core.Response;
+import org.apache.syncope.common.lib.to.JobTO;
import org.apache.syncope.common.lib.to.NotificationTO;
+import org.apache.syncope.common.lib.types.JobAction;
import org.apache.syncope.common.rest.api.RESTHeaders;
import org.apache.syncope.common.rest.api.service.NotificationService;
import org.apache.syncope.core.logic.NotificationLogic;
@@ -62,4 +64,14 @@ public class NotificationServiceImpl extends AbstractServiceImpl implements Noti
public void delete(final Long key) {
logic.delete(key);
}
+
+ @Override
+ public JobTO getJob() {
+ return logic.getJob();
+ }
+
+ @Override
+ public void actionJob(final JobAction action) {
+ logic.actionJob(action);
+ }
}
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/ReportServiceImpl.java
----------------------------------------------------------------------
diff --git a/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/ReportServiceImpl.java b/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/ReportServiceImpl.java
index 44fde36..d72e484 100644
--- a/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/ReportServiceImpl.java
+++ b/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/ReportServiceImpl.java
@@ -26,10 +26,10 @@ import javax.ws.rs.core.HttpHeaders;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.StreamingOutput;
import org.apache.syncope.common.lib.to.BulkActionResult;
+import org.apache.syncope.common.lib.to.JobTO;
import org.apache.syncope.common.lib.to.ReportExecTO;
import org.apache.syncope.common.lib.to.ReportTO;
import org.apache.syncope.common.lib.types.JobAction;
-import org.apache.syncope.common.lib.types.JobStatusType;
import org.apache.syncope.common.lib.types.ReportExecExportFormat;
import org.apache.syncope.common.rest.api.RESTHeaders;
import org.apache.syncope.common.rest.api.beans.BulkExecDeleteQuery;
@@ -98,6 +98,11 @@ public class ReportServiceImpl extends AbstractServiceImpl implements ReportServ
}
@Override
+ public List<ReportExecTO> listRecentExecutions(final int size) {
+ return logic.listRecentExecutions(size);
+ }
+
+ @Override
public void deleteExecution(final Long executionKey) {
logic.deleteExecution(executionKey);
}
@@ -113,8 +118,8 @@ public class ReportServiceImpl extends AbstractServiceImpl implements ReportServ
}
@Override
- public List<ReportExecTO> listJobs(final JobStatusType type) {
- return logic.listJobs(type, ReportExecTO.class);
+ public List<JobTO> listJobs(final int max) {
+ return logic.listJobs(max);
}
@Override
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/TaskServiceImpl.java
----------------------------------------------------------------------
diff --git a/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/TaskServiceImpl.java b/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/TaskServiceImpl.java
index 495fff8..0a1bf17 100644
--- a/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/TaskServiceImpl.java
+++ b/core/rest-cxf/src/main/java/org/apache/syncope/core/rest/cxf/service/TaskServiceImpl.java
@@ -25,13 +25,13 @@ import javax.ws.rs.core.Response;
import org.apache.syncope.common.lib.to.AbstractTaskTO;
import org.apache.syncope.common.lib.to.BulkAction;
import org.apache.syncope.common.lib.to.BulkActionResult;
+import org.apache.syncope.common.lib.to.JobTO;
import org.apache.syncope.common.lib.to.PagedResult;
import org.apache.syncope.common.lib.to.PushTaskTO;
import org.apache.syncope.common.lib.to.SchedTaskTO;
import org.apache.syncope.common.lib.to.SyncTaskTO;
import org.apache.syncope.common.lib.to.TaskExecTO;
import org.apache.syncope.common.lib.types.JobAction;
-import org.apache.syncope.common.lib.types.JobStatusType;
import org.apache.syncope.common.rest.api.RESTHeaders;
import org.apache.syncope.common.rest.api.beans.BulkExecDeleteQuery;
import org.apache.syncope.common.rest.api.beans.ExecuteQuery;
@@ -123,6 +123,11 @@ public class TaskServiceImpl extends AbstractServiceImpl implements TaskService
}
@Override
+ public List<TaskExecTO> listRecentExecutions(final int max) {
+ return logic.listRecentExecutions(max);
+ }
+
+ @Override
public void deleteExecution(final Long executionKey) {
logic.deleteExecution(executionKey);
}
@@ -186,8 +191,8 @@ public class TaskServiceImpl extends AbstractServiceImpl implements TaskService
}
@Override
- public List<TaskExecTO> listJobs(final JobStatusType type) {
- return logic.listJobs(type, TaskExecTO.class);
+ public List<JobTO> listJobs(final int max) {
+ return logic.listJobs(max);
}
@Override
http://git-wip-us.apache.org/repos/asf/syncope/blob/15582e4f/fit/core-reference/src/test/java/org/apache/syncope/fit/core/SchedTaskITCase.java
----------------------------------------------------------------------
diff --git a/fit/core-reference/src/test/java/org/apache/syncope/fit/core/SchedTaskITCase.java b/fit/core-reference/src/test/java/org/apache/syncope/fit/core/SchedTaskITCase.java
index 0b67502..20768fd 100644
--- a/fit/core-reference/src/test/java/org/apache/syncope/fit/core/SchedTaskITCase.java
+++ b/fit/core-reference/src/test/java/org/apache/syncope/fit/core/SchedTaskITCase.java
@@ -29,15 +29,17 @@ import java.util.Date;
import java.util.List;
import java.util.Set;
import javax.ws.rs.core.Response;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.collections4.Predicate;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.syncope.common.lib.to.AbstractTaskTO;
+import org.apache.syncope.common.lib.to.JobTO;
import org.apache.syncope.common.lib.to.PagedResult;
import org.apache.syncope.common.lib.to.PushTaskTO;
import org.apache.syncope.common.lib.to.SchedTaskTO;
import org.apache.syncope.common.lib.to.SyncTaskTO;
import org.apache.syncope.common.lib.to.TaskExecTO;
import org.apache.syncope.common.lib.types.JobAction;
-import org.apache.syncope.common.lib.types.JobStatusType;
import org.apache.syncope.common.lib.types.TaskType;
import org.apache.syncope.common.rest.api.beans.ExecuteQuery;
import org.apache.syncope.common.rest.api.beans.TaskExecQuery;
@@ -159,8 +161,8 @@ public class SchedTaskITCase extends AbstractTaskITCase {
@Test
public void issueSYNCOPE660() {
- List<TaskExecTO> list = taskService.listJobs(JobStatusType.ALL);
- int old_size = list.size();
+ List<JobTO> jobs = taskService.listJobs(50);
+ int old_size = jobs.size();
SchedTaskTO task = new SchedTaskTO();
task.setName("issueSYNCOPE660");
@@ -170,8 +172,8 @@ public class SchedTaskITCase extends AbstractTaskITCase {
Response response = taskService.create(task);
task = getObject(response.getLocation(), TaskService.class, SchedTaskTO.class);
- list = taskService.listJobs(JobStatusType.ALL);
- assertEquals(old_size + 1, list.size());
+ jobs = taskService.listJobs(50);
+ assertEquals(old_size + 1, jobs.size());
taskService.actionJob(task.getKey(), JobAction.START);
@@ -184,14 +186,19 @@ public class SchedTaskITCase extends AbstractTaskITCase {
// ignore
}
- list = taskService.listJobs(JobStatusType.RUNNING);
+ jobs = taskService.listJobs(50);
+ CollectionUtils.filter(jobs, new Predicate<JobTO>() {
- assertNotNull(list);
+ @Override
+ public boolean evaluate(final JobTO job) {
+ return job.isRunning();
+ }
+ });
i++;
- } while (list.size() < 1 && i < maxit);
+ } while (jobs.size() < 1 && i < maxit);
- assertEquals(1, list.size());
- assertEquals(task.getKey(), list.get(0).getTask(), 0);
+ assertEquals(1, jobs.size());
+ assertEquals(task.getKey(), jobs.get(0).getReferenceKey(), 0);
taskService.actionJob(task.getKey(), JobAction.STOP);
@@ -204,12 +211,17 @@ public class SchedTaskITCase extends AbstractTaskITCase {
// ignore
}
- list = taskService.listJobs(JobStatusType.RUNNING);
+ jobs = taskService.listJobs(50);
+ CollectionUtils.filter(jobs, new Predicate<JobTO>() {
- assertNotNull(list);
+ @Override
+ public boolean evaluate(final JobTO job) {
+ return job.isRunning();
+ }
+ });
i++;
- } while (list.size() >= 1 && i < maxit);
+ } while (jobs.size() >= 1 && i < maxit);
- assertTrue(list.isEmpty());
+ assertTrue(jobs.isEmpty());
}
}