package com.atlassian.scheduler.quartz1;
   
   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 com.atlassian.scheduler.core.AbstractSchedulerService;
  import com.atlassian.scheduler.core.JobLauncher;
  import com.atlassian.scheduler.core.spi.RunDetailsDao;
  import com.atlassian.scheduler.core.spi.SchedulerServiceConfiguration;
  import com.atlassian.scheduler.quartz1.spi.Quartz1SchedulerConfiguration;
  import com.atlassian.scheduler.status.JobDetails;
  import com.google.common.collect.ImmutableList;
  import com.google.common.collect.ImmutableSet;
  import org.quartz.Scheduler;
  import org.quartz.Trigger;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import javax.annotation.Nonnull;
  import javax.annotation.Nullable;
  import java.util.Date;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Set;
  
  import static com.atlassian.scheduler.config.RunMode.RUN_LOCALLY;
  import static com.atlassian.scheduler.config.RunMode.RUN_ONCE_PER_CLUSTER;
  import static com.atlassian.scheduler.quartz1.Quartz1SchedulerFacade.createClustered;
  import static com.atlassian.scheduler.quartz1.Quartz1SchedulerFacade.createLocal;
  import static com.atlassian.util.concurrent.Assertions.notNull;
  import static com.google.common.collect.Lists.newArrayList;
  import static com.google.common.collect.Sets.newHashSet;
  import static java.util.Collections.sort;
  
  /**
   * Quartz 1.x implementation of a {@link SchedulerService}.
   * <ul>
   * <li>Job runner keys are mapped to Quartz {@code Job} names, with the {@code Job} being created or
   * destroyed automatically based on whether or not it has any existing {@code Trigger}s.</li>
   * <li>All Quartz {@code Job}s use {@link Quartz1Job}, which immediately delegates to {@link JobLauncher}.</li>
   * <li>Job ids are mapped to Quartz {@code Trigger} names.</li>
   * <li>The parameters map is serialized to a {@code byte[]} and stored in the {@code JobDataMap}
   * for the Quartz {@code Trigger}.</li>
   * </ul>
   */
  public class Quartz1SchedulerService extends AbstractSchedulerService {
      private static final Logger LOG = LoggerFactory.getLogger(Quartz1SchedulerService.class);
  
      private final Quartz1SchedulerFacade localJobs;
      private final Quartz1SchedulerFacade clusteredJobs;
      private final Quartz1TriggerFactory triggerFactory;
      private final Quartz1JobDetailsFactory jobDetailsFactory;
  
  
      @SuppressWarnings("unused")
      public Quartz1SchedulerService(final RunDetailsDao runDetailsDao, final Quartz1SchedulerConfiguration config)
              throws SchedulerServiceException {
          super(runDetailsDao);
          this.localJobs = createLocal(this, config);
          this.clusteredJobs = createClustered(this, config);
          this.triggerFactory = new Quartz1TriggerFactory(config, getParameterMapSerializer());
          this.jobDetailsFactory = new Quartz1JobDetailsFactory(this);
      }
  
      @SuppressWarnings("unused")
      public Quartz1SchedulerService(final RunDetailsDao runDetailsDao, final SchedulerServiceConfiguration config,
                                     final Scheduler localScheduler, final Scheduler clusteredScheduler)
              throws SchedulerServiceException {
          super(runDetailsDao);
          this.localJobs = createLocal(this, localScheduler);
          this.clusteredJobs = createClustered(this, clusteredScheduler);
          this.triggerFactory = new Quartz1TriggerFactory(config, getParameterMapSerializer());
          this.jobDetailsFactory = new Quartz1JobDetailsFactory(this);
      }
  
  
      @Override
      public void scheduleJob(final JobId jobId, final JobConfig jobConfig)
              throws SchedulerServiceException {
          try {
              notNull("jobConfig", jobConfig);
              LOG.debug("scheduleJob: {}: {}", jobId, jobConfig);
  
              localJobs.unscheduleJob(jobId);
              clusteredJobs.unscheduleJob(jobId);
  
              final Quartz1SchedulerFacade facade = getFacade(jobConfig.getRunMode());
              final JobRunnerKey jobRunnerKey = jobConfig.getJobRunnerKey();
              final Trigger trigger = triggerFactory.buildTrigger(jobId, jobConfig);
              facade.scheduleJob(jobRunnerKey, trigger);
          } catch (SchedulerRuntimeException sre) {
              throw checked(sre);
          }
      }
 
 
     @Override
     public void unscheduleJob(final JobId jobId) {
         // Deliberately avoiding a short-circuit; we want to remove from both
         boolean found = localJobs.unscheduleJob(jobId);
         found |= clusteredJobs.unscheduleJob(jobId);
         if (found) {
             LOG.debug("unscheduleJob: {}", jobId);
         } else {
             LOG.debug("unscheduleJob for non-existent jobId: {}", jobId);
         }
     }
 
     @Nullable
     @Override
     public Date calculateNextRunTime(Schedule schedule) throws SchedulerServiceException {
         final Trigger trigger = triggerFactory.buildTrigger(schedule);
         return trigger.getFireTimeAfter(new Date());
     }
 
     @Nullable
     @Override
     public JobDetails getJobDetails(final JobId jobId) {
         JobDetails jobDetails = getJobDetails(clusteredJobs, jobId, RUN_ONCE_PER_CLUSTER);
         if (jobDetails == null) {
             jobDetails = getJobDetails(localJobs, jobId, RUN_LOCALLY);
         }
         return jobDetails;
     }
 
     @Nullable
     private JobDetails getJobDetails(Quartz1SchedulerFacade facade, JobId jobId, RunMode runMode) {
         final Trigger trigger = facade.getTrigger(jobId);
         return (trigger != null) ? jobDetailsFactory.buildJobDetails(jobId, trigger, runMode) : null;
     }
 
     @Nonnull
     @Override
     public Set<JobRunnerKey> getJobRunnerKeysForAllScheduledJobs() {
         final Set<JobRunnerKey> jobRunnerKeys = new HashSet<JobRunnerKey>(localJobs.getJobRunnerKeys());
         jobRunnerKeys.addAll(clusteredJobs.getJobRunnerKeys());
         return ImmutableSet.copyOf(jobRunnerKeys);
     }
 
     @Nonnull
     @Override
     public List<JobDetails> getJobsByJobRunnerKey(final JobRunnerKey jobRunnerKey) {
         final UniqueJobDetailsCollector collector = new UniqueJobDetailsCollector();
         collector.collect(RUN_ONCE_PER_CLUSTER, clusteredJobs.getTriggersOfJob(jobRunnerKey));
         collector.collect(RUN_LOCALLY, localJobs.getTriggersOfJob(jobRunnerKey));
         return collector.getResults();
     }
 
     /**
      * Returns the actual Quartz scheduler providing {@link RunMode#RUN_LOCALLY local} scheduling.  This
      * method will be removed in a future release and its use should be avoided if at all possible.  It is
      * provided only because Atlassian applications have historically exposed Quartz 1.x to plugins and
      * may need to continue to do so as part of the migration to the {@code atlassian-scheduler} library.
      *
      * @return the actual Quartz scheduler providing {@link RunMode#RUN_LOCALLY local} scheduling
      * @see #getClusteredQuartzScheduler()
      * @deprecated Provided as a last resort only.  Avoid accessing Quartz directly if possible.  Since v1.0
      */
     @Deprecated
     public Scheduler getLocalQuartzScheduler() {
         return localJobs.getQuartz();
     }
 
     /**
      * Returns the actual Quartz scheduler providing {@link RunMode#RUN_ONCE_PER_CLUSTER clustered} scheduling.  This
      * method will be removed in a future release and its use should be avoided if at all possible.  It is
      * provided only because Atlassian applications have historically exposed Quartz 1.x to plugins and
      * may need to continue to do so as part of the migration to the {@code atlassian-scheduler} library.
      *
      * @return the actual Quartz scheduler providing {@link RunMode#RUN_ONCE_PER_CLUSTER clustered} scheduling
      * @see #getLocalQuartzScheduler()
      * @deprecated Provided as a last resort only.  Avoid accessing Quartz directly if possible.  Since v1.0
      */
     @Deprecated
     public Scheduler getClusteredQuartzScheduler() {
         return clusteredJobs.getQuartz();
     }
 
     /**
      * Start the threads associated with each quartz scheduler. It is the responsibility of the underlying
      * JobStores to determine whether or not to trigger jobs which should have run while the scheduler was
      * in standby mode. This is usually controlled by a misfire threshold on the JobStore implementation
      */
     @Override
     protected void startImpl() throws SchedulerServiceException {
         boolean abort = true;
         localJobs.start();
         try {
             clusteredJobs.start();
             abort = false;
         } finally {
             if (abort) {
                 localJobs.standby();
             }
         }
     }
 
     /**
      * Stop the threads associated with each quartz scheduler
      */
     @Override
     protected void standbyImpl() throws SchedulerServiceException {
         try {
             localJobs.standby();
         } finally {
             clusteredJobs.standby();
         }
     }
 
     @Override
     protected void shutdownImpl() {
         try {
             localJobs.shutdown();
         } finally {
             clusteredJobs.shutdown();
         }
     }
 
     private Quartz1SchedulerFacade getFacade(RunMode runMode) {
         switch (runMode) {
             case RUN_LOCALLY:
                 return localJobs;
             case RUN_ONCE_PER_CLUSTER:
                 return clusteredJobs;
         }
         throw new IllegalArgumentException("runMode=" + runMode);
     }
 
 
     /**
      * Since it is possible for another node in the cluster to register the same jobId for running clustered
      * when we had it set to run locally, deal with this by pretending the local one is not there.
      */
     class UniqueJobDetailsCollector {
         final Set<String> jobIdsSeen = newHashSet();
         final List<JobDetails> jobs = newArrayList();
 
         void collect(RunMode runMode, final Trigger[] triggers) {
             for (Trigger trigger : triggers) {
                 final String jobId = trigger.getName();
                 if (jobIdsSeen.add(jobId)) {
                     try {
                         jobs.add(jobDetailsFactory.buildJobDetails(JobId.of(jobId), trigger, runMode));
                     } catch (SchedulerRuntimeException sre) {
                         LOG.debug("Unable to reconstruct log details for jobId '{}'", jobId, sre);
                     }
                 }
             }
         }
 
         List<JobDetails> getResults() {
             sort(jobs, BY_JOB_ID);
             return ImmutableList.copyOf(jobs);
         }
     }
 
 }