package com.atlassian.sal.core.scheduling;
   
   import com.atlassian.sal.api.scheduling.PluginJob;
   import com.atlassian.sal.api.scheduling.PluginScheduler;
   import com.atlassian.scheduler.JobRunner;
   import com.atlassian.scheduler.JobRunnerRequest;
   import com.atlassian.scheduler.JobRunnerResponse;
   import com.atlassian.scheduler.SchedulerRuntimeException;
   import com.atlassian.scheduler.SchedulerService;
  import com.atlassian.scheduler.SchedulerServiceException;
  import com.atlassian.scheduler.config.JobConfig;
  import com.atlassian.scheduler.config.JobId;
  import com.atlassian.scheduler.config.JobRunnerKey;
  import com.atlassian.scheduler.config.RunMode;
  import com.atlassian.scheduler.config.Schedule;
  import org.springframework.beans.factory.DisposableBean;
  import org.springframework.beans.factory.InitializingBean;
  
  import javax.annotation.Nonnull;
  import javax.annotation.Nullable;
  import java.util.Date;
  import java.util.Map;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ConcurrentMap;
  
  import static com.atlassian.scheduler.JobRunnerResponse.aborted;
  import static com.atlassian.util.concurrent.Assertions.notNull;
  
  /**
   * A plugin scheduler that is backed by the atlassian-scheduler library.
   *
   * @since v3.0.0
   */
  public class DefaultPluginScheduler implements PluginScheduler, InitializingBean, DisposableBean {
      static final JobRunnerKey JOB_RUNNER_KEY = JobRunnerKey.of(DefaultPluginScheduler.class.getName());
  
      // Must keep them locally because the scheduler service only accepts Serializable args, not arbitrary Objects
      private final ConcurrentMap<JobId, JobDescriptor> descriptors = new ConcurrentHashMap<JobId, JobDescriptor>();
  
      private final SchedulerService schedulerService;
  
      public DefaultPluginScheduler(SchedulerService schedulerService) {
          this.schedulerService = schedulerService;
  
          // Registering here would allow "this" to escape the thread before it is properly constructed.
          // See Java Concurrency in Practice, section 3.2.1.
      }
  
      @Override
      public void scheduleJob(String jobKey, Class<? extends PluginJob> jobClass, Map<String, Object> jobDataMap, Date startTime, long repeatInterval) {
          final JobId jobId = toJobId(jobKey);
          descriptors.put(jobId, new JobDescriptor(jobClass, jobDataMap));
          final JobConfig jobConfig = JobConfig.forJobRunnerKey(JOB_RUNNER_KEY)
                  .withRunMode(RunMode.RUN_LOCALLY)
                  .withSchedule(Schedule.forInterval(repeatInterval, startTime));
          try {
              schedulerService.scheduleJob(jobId, jobConfig);
          } catch (SchedulerServiceException sse) {
              throw new SchedulerRuntimeException(sse.getMessage(), sse);
          }
      }
  
      @Override
      public void unscheduleJob(String jobKey) {
          final JobId jobId = toJobId(jobKey);
  
          // Do the real unschedule regardless
          schedulerService.unscheduleJob(jobId);
  
          // Why not be idempotent?  Because unfortunately the API demands this behaviour. :(
          if (descriptors.remove(jobId) == null) {
              throw new IllegalArgumentException("Error unscheduling job. Job '" + jobKey + "' is not scheduled.");
          }
      }
  
      @Nonnull
      JobRunnerResponse runJobImpl(JobRunnerRequest jobRunnerRequest) {
          final JobDescriptor descriptor = descriptors.get(jobRunnerRequest.getJobId());
          if (descriptor == null) {
              return aborted("Job descriptor not found");
          }
          return descriptor.runJob();
      }
  
      @Override
      public void afterPropertiesSet() {
          // Register the job runner here, instead
          schedulerService.registerJobRunner(JOB_RUNNER_KEY, new JobRunner() {
              @Nullable
              @Override
              public JobRunnerResponse runJob(JobRunnerRequest jobRunnerRequest) {
                  return runJobImpl(jobRunnerRequest);
              }
          });
      }
  
      @Override
      public void destroy() {
          schedulerService.unregisterJobRunner(JOB_RUNNER_KEY);
     }
 
     static JobId toJobId(String jobKey) {
         return JobId.of(DefaultPluginScheduler.class.getSimpleName() + ':' + jobKey);
     }
 
 
     static class JobDescriptor {
         final Class<? extends PluginJob> jobClass;
         final Map<String, Object> jobDataMap;
 
         @SuppressWarnings("AssignmentToCollectionOrArrayFieldFromParameter")
             // Required for compatibility
         JobDescriptor(final Class<? extends PluginJob> jobClass, final Map<String, Object> jobDataMap) {
             this.jobClass = notNull("jobClass", jobClass);
             this.jobDataMap = jobDataMap;
         }
 
         @Nonnull
         JobRunnerResponse runJob() {
             final PluginJob job;
             try {
                 job = jobClass.newInstance();
             } catch (InstantiationException e) {
                 return aborted(e.toString());
             } catch (IllegalAccessException e) {
                 return aborted(e.toString());
             }
 
             job.execute(jobDataMap);
             return JobRunnerResponse.success();
         }
     }
 }