package com.atlassian.sal.core.scheduling;
   
   import com.atlassian.sal.api.scheduling.PluginJob;
   import com.atlassian.sal.api.scheduling.PluginScheduler;
   
   import java.util.Date;
   import java.util.Map;
   import java.util.concurrent.ConcurrentHashMap;
   import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.concurrent.ScheduledExecutorService;
  import java.util.concurrent.TimeUnit;
  
  /**
   * Plugin scheduler that uses java.util.concurrent.Executor.
   */
  public class ExecutorPluginScheduler implements PluginScheduler {
  
      public static final int DEFAULT_POOL_SIZE = 5;
      private final ScheduledExecutorService jobExecutor;
      private final Map<String, Future<?>> jobs;
  
      public ExecutorPluginScheduler() {
          this(Executors.newScheduledThreadPool(DEFAULT_POOL_SIZE));
      }
  
      public ExecutorPluginScheduler(ScheduledExecutorService executor) {
          this.jobExecutor = executor;
          this.jobs = new ConcurrentHashMap<>();
      }
  
      @Override
      public synchronized void scheduleJob(String jobKey, Class<? extends PluginJob> jobClass, Map<String, Object> jobDataMap, Date startTime, long repeatInterval) {
          final Future<?> job = jobs.get(jobKey);
          if (job != null) {
              cancelJob(job);
          }
  
          jobs.put(jobKey, jobExecutor.scheduleAtFixedRate(new Job(jobClass, jobDataMap), getDelay(startTime), repeatInterval, TimeUnit.MILLISECONDS));
      }
  
      @Override
      public synchronized void unscheduleJob(String jobKey) {
          final Future<?> job = jobs.remove(jobKey);
          if (job != null) {
              cancelJob(job);
          } else {
              throw new IllegalArgumentException("Attempted to unschedule unknown job: " + jobKey);
          }
      }
  
      protected void cancelJob(Future<?> job) {
          job.cancel(false);
      }
  
      private long getDelay(Date startTime) {
          final long time = startTime.getTime() - System.currentTimeMillis();
          return time > 0 ? time : 0;
      }
  
      private static class Job implements Runnable {
  
          private final Class<? extends PluginJob> jobClass;
          private final Map<String, Object> jobDataMap;
  
          private Job(Class<? extends PluginJob> jobClass, Map<String, Object> jobDataMap) {
              this.jobClass = jobClass;
              this.jobDataMap = jobDataMap;
          }
  
          @Override
          public void run() {
              PluginJob job;
              try {
                  job = jobClass.newInstance();
              } catch (final InstantiationException ie) {
                  throw new IllegalStateException("Error instantiating job", ie);
              } catch (final IllegalAccessException iae) {
                  throw new IllegalStateException("Cannot access job class", iae);
              }
              job.execute(jobDataMap);
          }
      }
  
  }