Merge pull request #207 from wttech/script-scheduler-job-consumer

Script scheduler as job consumer

Author: krystian-panek-vmltech

core/src/main/java/dev/vml/es/acm/core/script/AutomaticScriptScheduler.java renamed to core/src/main/java/dev/vml/es/acm/core/script/ScriptScheduler.java

@@ -1,6 +1,5 @@
 package dev.vml.es.acm.core.script;
 
-import dev.vml.es.acm.core.AcmConstants;
 import dev.vml.es.acm.core.code.*;
 import dev.vml.es.acm.core.code.schedule.BootSchedule;
 import dev.vml.es.acm.core.code.schedule.CronSchedule;
@@ -15,19 +14,27 @@
 import dev.vml.es.acm.core.repo.Repo;
 import dev.vml.es.acm.core.util.ChecksumUtils;
 import dev.vml.es.acm.core.util.ResolverUtils;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.stream.Stream;
+
 import org.apache.commons.lang3.StringUtils;
 import org.apache.sling.api.resource.LoginException;
 import org.apache.sling.api.resource.ResourceResolver;
 import org.apache.sling.api.resource.ResourceResolverFactory;
 import org.apache.sling.api.resource.observation.ResourceChange;
 import org.apache.sling.api.resource.observation.ResourceChangeListener;
-import org.apache.sling.commons.scheduler.Job;
-import org.apache.sling.commons.scheduler.ScheduleOptions;
-import org.apache.sling.commons.scheduler.Scheduler;
+import org.apache.sling.event.jobs.Job;
+import org.apache.sling.event.jobs.JobBuilder;
+import org.apache.sling.event.jobs.JobManager;
+import org.apache.sling.event.jobs.ScheduledJobInfo;
+import org.apache.sling.event.jobs.consumer.JobConsumer;
 import org.osgi.service.component.annotations.*;
 import org.osgi.service.metatype.annotations.AttributeDefinition;
 import org.osgi.service.metatype.annotations.Designate;
@@ -36,26 +43,47 @@
 import org.slf4j.LoggerFactory;
 
 @Component(
- service = {ResourceChangeListener.class, EventListener.class},
+ service = {ResourceChangeListener.class, EventListener.class, JobConsumer.class},
  immediate = true,
  property = {
  ResourceChangeListener.PATHS + "=glob:" + ScriptRepository.ROOT + "/automatic/**/*.groovy",
  ResourceChangeListener.CHANGES + "=ADDED",
  ResourceChangeListener.CHANGES + "=CHANGED",
- ResourceChangeListener.CHANGES + "=REMOVED"
+ ResourceChangeListener.CHANGES + "=REMOVED",
+ JobConsumer.PROPERTY_TOPICS + "=" + ScriptScheduler.JOB_TOPIC
  })
-@Designate(ocd = AutomaticScriptScheduler.Config.class)
-public class AutomaticScriptScheduler implements ResourceChangeListener, EventListener {
+@Designate(ocd = ScriptScheduler.Config.class)
+public class ScriptScheduler implements ResourceChangeListener, EventListener, JobConsumer {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ScriptScheduler.class);
+
+ public static final String JOB_TOPIC = "dev/vml/es/acm/ScriptScheduler";
+
+ public static final String JOB_PROP_TYPE = "type";
 
- private static final Logger LOG = LoggerFactory.getLogger(AutomaticScriptScheduler.class);
+ public static final String JOB_PROP_SCRIPT_PATH = "scriptPath";
 
- private static final String BOOT_JOB_NAME = AcmConstants.CODE + "-boot";
+ public enum JobType {
+ BOOT,
+ CRON;
+
+ public static JobType of(String value) {
+ return Arrays.stream(values())
+ .filter(v -> StringUtils.equalsIgnoreCase(v.name(), value))
+ .findFirst()
+ .orElseThrow(
+ () -> new IllegalArgumentException("Script scheduler job type is unsupported: " + value));
+ }
+ }
 
  @ObjectClassDefinition(
- name = "AEM Content Manager - Automatic Script Scheduler",
+ name = "AEM Content Manager - Script Scheduler",
  description = "Schedules automatic scripts on instance up and script changes")
  public @interface Config {
 
+ @AttributeDefinition(name = "Boot Delay", description = "Time in milliseconds to wait before booting scripts")
+ long bootDelay() default 1000 * 10; // 10 seconds
+
  @AttributeDefinition(
  name = "User Impersonation ID",
  description =
@@ -81,9 +109,7 @@ public class AutomaticScriptScheduler implements ResourceChangeListener, EventLi
 
  private Boolean instanceReady;
 
- private final Map<String, String> booted = new ConcurrentHashMap<>();
-
- private final List<String> scheduled = new CopyOnWriteArrayList<>();
+ private final Map<String, String> bootedScripts = new ConcurrentHashMap<>();
 
  @Reference
  private ResourceResolverFactory resourceResolverFactory;
@@ -95,7 +121,7 @@ public class AutomaticScriptScheduler implements ResourceChangeListener, EventLi
  private Executor executor;
 
  @Reference
- private Scheduler scheduler;
+ private JobManager jobManager;
 
  @Reference
  private HealthChecker healthChecker;
@@ -123,7 +149,7 @@ protected void modify(Config config) {
  protected void deactivate() {
  unscheduleBoot();
  unscheduleScripts();
- booted.clear();
+ bootedScripts.clear();
  instanceReady = null;
  }
 
@@ -137,8 +163,19 @@ public void onChange(List<ResourceChange> changes) {
  @Override
  public void onEvent(Event event) {
  EventType eventType = EventType.of(event.getName()).orElse(null);
- if (eventType == EventType.SCRIPT_SCHEDULER_BOOT) {
- bootOnDemand();
+ if (eventType == null) {
+ return;
+ }
+ switch (eventType) {
+ case SCRIPT_SCHEDULER_BOOT:
+ bootOnDemand();
+ break;
+ case EXECUTOR_RESET:
+ reset();
+ break;
+ default:
+ // ignore else
+ break;
  }
  }
 
@@ -149,7 +186,6 @@ public void bootOnDemand() {
  LOG.info("Automatic scripts booting on demand - job scheduled");
  }
 
- // TODO on AEMaaCS scheduler refuses to schedule job during activate
  private void bootWhenInstanceUp() {
  LOG.info("Automatic scripts booting on instance up - job scheduling");
  unscheduleBoot();
@@ -164,19 +200,35 @@ private void bootWhenScriptsChanged() {
  LOG.info("Automatic scripts booting on script changes - job scheduled");
  }
 
+ @SuppressWarnings("unchecked")
  private void unscheduleBoot() {
- scheduler.unschedule(BOOT_JOB_NAME);
+ Collection<ScheduledJobInfo> jobInfos = jobManager.getScheduledJobs(
+ JOB_TOPIC, -1, Collections.singletonMap(JOB_PROP_TYPE, JobType.BOOT.name()));
+ for (ScheduledJobInfo jobInfo : jobInfos) {
+ jobInfo.unschedule();
+ }
  }
 
  private void scheduleBoot() {
- scheduler.schedule(bootJob(), configureScheduleOptions(BOOT_JOB_NAME, scheduler.NOW()));
+ JobBuilder jobBuilder = jobManager.createJob(JOB_TOPIC);
+ jobBuilder.properties(Collections.singletonMap(JOB_PROP_TYPE, JobType.BOOT.name()));
+ JobBuilder.ScheduleBuilder scheduleBuilder = jobBuilder.schedule();
+ scheduleBuilder.at(new Date(System.currentTimeMillis() + config.bootDelay()));
+ scheduleBuilder.add();
  }
 
- private ScheduleOptions configureScheduleOptions(String name, ScheduleOptions options) {
- options.name(name);
- options.canRunConcurrently(false);
- options.onSingleInstanceOnly(true);
- return options;
+ @Override
+ public JobResult process(Job job) {
+ switch (JobType.of(job.getProperty(JOB_PROP_TYPE, String.class))) {
+ case BOOT:
+ bootJob();
+ break;
+ case CRON:
+ String scriptPath = job.getProperty(JOB_PROP_SCRIPT_PATH, String.class);
+ cronJob(scriptPath);
+ break;
+ }
+ return JobResult.OK;
  }
 
  private ScheduleResult determineSchedule(Script script, ResourceResolver resourceResolver) {
@@ -186,18 +238,16 @@ private ScheduleResult determineSchedule(Script script, ResourceResolver resourc
  }
  }
 
- private Job bootJob() {
- return context -> {
- LOG.info("Automatic scripts booting - job started");
- unscheduleScripts();
- if (awaitInstanceHealthy(
- "Automatic scripts queueing and scheduling",
- config.healthCheckRetryCountBoot(),
- config.healthCheckRetryInterval())) {
- queueAndScheduleScripts();
- }
- LOG.info("Automatic scripts booting - job finished");
- };
+ private void bootJob() {
+ LOG.info("Automatic scripts booting - job started");
+ unscheduleScripts();
+ if (awaitInstanceHealthy(
+ "Automatic scripts queueing and scheduling",
+ config.healthCheckRetryCountBoot(),
+ config.healthCheckRetryInterval())) {
+ queueAndScheduleScripts();
+ }
+ LOG.info("Automatic scripts booting - job finished");
  }
 
  private boolean checkInstanceReady() {
@@ -216,15 +266,13 @@ private boolean checkInstanceReady() {
  return instanceReady;
  }
 
+ @SuppressWarnings("unchecked")
  private void unscheduleScripts() {
- for (String scriptPath : scheduled) {
- try {
- scheduler.unschedule(scriptPath);
- } catch (Exception e) {
- LOG.error("Cron schedule script '{}' cannot be unscheduled!", scriptPath, e);
- }
+ Collection<ScheduledJobInfo> jobInfos = jobManager.getScheduledJobs(
+ JOB_TOPIC, -1, Collections.singletonMap(JOB_PROP_TYPE, JobType.CRON.name()));
+ for (ScheduledJobInfo jobInfo : jobInfos) {
+ jobInfo.unschedule();
  }
- scheduled.clear();
  }
 
  private void queueAndScheduleScripts() {
@@ -257,11 +305,11 @@ private void queueAndScheduleScripts() {
 
  private void queueBootScript(Script script, ResourceResolver resourceResolver) {
  String checksum = ChecksumUtils.calculate(script.getContent());
- String previousChecksum = booted.get(script.getId());
+ String previousChecksum = bootedScripts.get(script.getId());
  if (previousChecksum == null || !StringUtils.equals(previousChecksum, checksum)) {
  if (checkScript(script, resourceResolver)) {
  queueScript(script);
- booted.put(script.getId(), checksum);
+ bootedScripts.put(script.getId(), checksum);
  LOG.info("Boot script '{}' queued", script.getId());
  } else {
  LOG.info("Boot script '{}' not eligible for queueing!", script.getId());
@@ -271,10 +319,14 @@ private void queueBootScript(Script script, ResourceResolver resourceResolver) {
 
  private void scheduleCronScript(Script script, CronSchedule schedule) {
  if (StringUtils.isNotBlank(schedule.getExpression())) {
- scheduler.schedule(
- cronJob(script.getPath()),
- configureScheduleOptions(script.getPath(), scheduler.EXPR(schedule.getExpression())));
- scheduled.add(script.getPath());
+ JobBuilder jobBuilder = jobManager.createJob(JOB_TOPIC);
+ Map<String, Object> properties = new HashMap<>();
+ properties.put(JOB_PROP_TYPE, JobType.CRON.name());
+ properties.put(JOB_PROP_SCRIPT_PATH, script.getPath());
+ jobBuilder.properties(properties);
+ JobBuilder.ScheduleBuilder scheduleBuilder = jobBuilder.schedule();
+ scheduleBuilder.cron(schedule.getExpression());
+ scheduleBuilder.add();
  LOG.info(
  "Cron schedule script '{}' scheduled with expression '{}'",
  script.getId(),
@@ -284,31 +336,29 @@ private void scheduleCronScript(Script script, CronSchedule schedule) {
  }
  }
 
- private Job cronJob(String scriptPath) {
- return context -> {
- LOG.info("Cron schedule script '{}' - job started", scriptPath);
- if (awaitInstanceHealthy(
- String.format("Cron schedule script '%s' queueing", scriptPath),
- config.healthCheckRetryCountCron(),
- config.healthCheckRetryInterval())) {
- try (ResourceResolver resourceResolver = ResolverUtils.contentResolver(resourceResolverFactory, null)) {
- ScriptRepository scriptRepository = new ScriptRepository(resourceResolver);
- Script script = scriptRepository.read(scriptPath).orElse(null);
- if (script == null) {
- LOG.error("Cron schedule script '{}' not found in repository!", scriptPath);
+ private void cronJob(String scriptPath) {
+ LOG.info("Cron schedule script '{}' - job started", scriptPath);
+ if (awaitInstanceHealthy(
+ String.format("Cron schedule script '%s' queueing", scriptPath),
+ config.healthCheckRetryCountCron(),
+ config.healthCheckRetryInterval())) {
+ try (ResourceResolver resourceResolver = ResolverUtils.contentResolver(resourceResolverFactory, null)) {
+ ScriptRepository scriptRepository = new ScriptRepository(resourceResolver);
+ Script script = scriptRepository.read(scriptPath).orElse(null);
+ if (script == null) {
+ LOG.error("Cron schedule script '{}' not found in repository!", scriptPath);
+ } else {
+ if (checkScript(script, resourceResolver)) {
+ queueScript(script);
  } else {
- if (checkScript(script, resourceResolver)) {
- queueScript(script);
- } else {
- LOG.info("Cron schedule script '{}' not eligible for queueing!", scriptPath);
- }
+ LOG.info("Cron schedule script '{}' not eligible for queueing!", scriptPath);
  }
- } catch (LoginException e) {
- LOG.error("Cannot access repository while queueing cron schedule script '{}'!", scriptPath, e);
  }
+ } catch (LoginException e) {
+ LOG.error("Cannot access repository while queueing cron schedule script '{}'!", scriptPath, e);
  }
-  LOG.info("Cron schedule script '{}' - job finished", scriptPath);
- };
+ }
+ LOG.info("Cron schedule script '{}' - job finished", scriptPath);
  }
 
  private boolean checkScript(Script script, ResourceResolver resourceResolver) {
@@ -387,4 +437,13 @@ private boolean awaitInstanceHealthy(String operation, long retryMaxCount, long
  }
  return true;
  }
+
+ public void reset() {
+ findJobs().forEach(job -> jobManager.removeJobById(job.getId()));
+ }
+
+ @SuppressWarnings("unchecked")
+ private Stream<Job> findJobs() {
+ return jobManager.findJobs(JobManager.QueryType.ALL, JOB_TOPIC, -1, Collections.emptyMap()).stream();
+ }
 }

ui.config/src/main/content/jcr_root/apps/acm-config/osgiconfig/config/org.apache.sling.event.jobs.QueueConfiguration-acmscriptscheduler.config

@@ -0,0 +1,8 @@
+queue.name="AEM\ Content\ Manager\ Script\ Scheduler"
+queue.topics=["dev/vml/es/acm/ScriptScheduler"]
+queue.priority="NORM"
+queue.type="ORDERED"
+queue.maxparallel="1"
+queue.retries="0"
+queue.retrydelay="2000"
+queue.keepJobs="false"

ui.frontend/src/components/ScriptAutomaticList.tsx

@@ -65,7 +65,7 @@ const ScriptAutomaticList: React.FC = () => {
  <ScriptsDeleteButton selectedKeys={selectedIds(selectedKeys)} onDelete={handleLoadScripts} />
  <ScriptsSyncButton selectedKeys={selectedIds(selectedKeys)} onSync={handleLoadScripts} />
  </Toggle>
- <Button variant="negative" isDisabled={appState.instanceSettings.type === InstanceType.CLOUD_CONTAINER} onPress={() => window.open(instanceOsgiServiceConfigUrl(InstanceOsgiServicePid.AUTOMATIC_SCRIPT_SCHEDULER), '_blank')}>
+ <Button variant="negative" isDisabled={appState.instanceSettings.type === InstanceType.CLOUD_CONTAINER} onPress={() => window.open(instanceOsgiServiceConfigUrl(InstanceOsgiServicePid.SCRIPT_SCHEDULER), '_blank')}>
  <Settings />
  <Text>Configure</Text>
  </Button>

ui.frontend/src/utils/api.types.ts

@@ -351,7 +351,7 @@ export enum InstanceType {
 export const instancePrefix = isProduction() ? '' : 'http://localhost:5502';
 
 export enum InstanceOsgiServicePid {
- AUTOMATIC_SCRIPT_SCHEDULER = 'dev.vml.es.acm.core.script.AutomaticScriptScheduler',
+ SCRIPT_SCHEDULER = 'dev.vml.es.acm.core.script.ScriptScheduler',
 
  CODE_EXECUTOR = 'dev.vml.es.acm.core.code.Executor',
  CODE_EXECUTION_QUEUE = 'dev.vml.es.acm.core.code.ExecutionQueue',