package com.atlassian.scheduler.core;
   
   import com.atlassian.scheduler.JobRunner;
   import com.atlassian.scheduler.JobRunnerRequest;
   import com.atlassian.scheduler.JobRunnerResponse;
   import com.atlassian.scheduler.SchedulerRuntimeException;
   import com.atlassian.scheduler.config.IntervalScheduleInfo;
   import com.atlassian.scheduler.config.JobConfig;
   import com.atlassian.scheduler.config.JobId;
  import com.atlassian.scheduler.config.RunMode;
  import com.atlassian.scheduler.core.impl.RunningJobImpl;
  import com.atlassian.scheduler.status.JobDetails;
  import com.atlassian.scheduler.status.RunOutcome;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import javax.annotation.Nonnull;
  import javax.annotation.Nullable;
  import java.util.Date;
  
  import static com.atlassian.scheduler.JobRunnerResponse.aborted;
  import static com.atlassian.scheduler.JobRunnerResponse.failed;
  import static com.atlassian.scheduler.status.RunOutcome.UNAVAILABLE;
  import static com.atlassian.util.concurrent.Assertions.notNull;
  
  /**
   * Scheduler implementations can (and should) use {@code JobLauncher} to invoke jobs when
   * it is time to run them.  It will do the necessary checks to make sure that the job runner
   * is registered, has the appropriate run mode, can reconstruct the job's parameter map, and
   * so on.  If everything checks out, it calls the job runner's
   * {@link JobRunner#runJob(JobRunnerRequest) runJob} method and records the
   * the resulting RunDetails using
   * {@link AbstractSchedulerService#addRunDetails(JobId, Date, RunOutcome, String) addRunDetails}.
   *
   * @since v1.0
   */
  public class JobLauncher {
      protected static final Logger LOG = LoggerFactory.getLogger(JobLauncher.class);
  
      protected final AbstractSchedulerService schedulerService;
      protected final RunMode schedulerRunMode;
      protected final Date firedAt;
      protected final JobId jobId;
  
      // Derived (in this order)
      private JobDetails jobDetails;
      private JobRunner jobRunner;
      private JobConfig jobConfig;
      private JobRunnerResponse response;
  
      /**
       * Creates a job launcher to handle the running of a scheduled job.
       *
       * @param schedulerService the scheduler that is invoking the job
       * @param schedulerRunMode the expected {@link RunMode run mode} for the jobs that this scheduler owns
       * @param firedAt          the time that the job was started, if known; may be {@code null}, in which case the
       *                         current time is used
       * @param jobId            the job ID of the job to run
       */
      public JobLauncher(final AbstractSchedulerService schedulerService, final RunMode schedulerRunMode,
                         @Nullable final Date firedAt, final JobId jobId) {
          this(schedulerService, schedulerRunMode, firedAt, jobId, null);
      }
  
      /**
       * Creates a job launcher to handle the running of a scheduled job.
       *
       * @param schedulerService the scheduler that is invoking the job
       * @param schedulerRunMode the expected {@link RunMode run mode} for the jobs that this scheduler owns
       * @param firedAt          the time that the job was started, if known; may be {@code null}, in which case the
       *                         current time is used
       * @param jobId            the job ID of the job to run
       * @param jobDetails       the already loaded job details, if they are available
       */
      public JobLauncher(final AbstractSchedulerService schedulerService, final RunMode schedulerRunMode,
                         @Nullable final Date firedAt, final JobId jobId, @Nullable final JobDetails jobDetails) {
          this.schedulerService = notNull("schedulerService", schedulerService);
          this.schedulerRunMode = notNull("schedulerRunMode", schedulerRunMode);
          this.firedAt = (firedAt != null) ? firedAt : new Date();
          this.jobId = notNull("jobId", jobId);
          this.jobDetails = jobDetails;
      }
  
      /**
       * Call this to validate the job, run it, and update its status.
       */
      public void launch() {
          LOG.debug("launch: {}: {}", schedulerRunMode, jobId);
          try {
              final JobRunnerResponse response = launchAndBuildResponse();
              schedulerService.addRunDetails(jobId, firedAt, response.getRunOutcome(), response.getMessage());
          } catch (JobRunnerNotRegisteredException ex) {
              LOG.debug("Scheduled job with ID '{}' is unavailable because its job runner is not registered: {}",
                      jobId, ex.getJobRunnerKey());
              schedulerService.addRunDetails(jobId, firedAt, UNAVAILABLE,
                      "Job runner key '" + ex.getJobRunnerKey() + "' is not registered");
          }
          deleteIfRunOnce();
      }
 
     @Nonnull
     private JobRunnerResponse launchAndBuildResponse() throws JobRunnerNotRegisteredException {
         try {
             response = validate();
             if (response == null) {
                 response = runJob();
             }
         } catch (RuntimeException ex) {
             LOG.error("Scheduled job with ID '{}' failed", jobId, ex);
             response = failed(ex);
         } catch (LinkageError err) {
             LOG.error("Scheduled job with ID '{}' failed due to binary incompatibilities", jobId, err);
             response = failed(err);
         }
         return response;
     }
 
     @Nonnull
     private JobRunnerResponse runJob() {
         final RunningJob job = new RunningJobImpl(firedAt, jobId, jobConfig);
 
         final RunningJob existing = schedulerService.enterJob(jobId, job);
         if (existing != null) {
             LOG.debug("Unable to start job {} because it is already running as {}", job, existing);
             return JobRunnerResponse.aborted("Already running");
         }
 
         schedulerService.preJob();
         final Thread thd = Thread.currentThread();
         final ClassLoader originalClassLoader = thd.getContextClassLoader();
         try {
             // SCHEDULER-11: Ensure that the Job runs with its own class loader set as the thread's CCL
             thd.setContextClassLoader(jobRunner.getClass().getClassLoader());
             final JobRunnerResponse response = jobRunner.runJob(job);
             return (response != null) ? response : JobRunnerResponse.success();
         } finally {
             thd.setContextClassLoader(originalClassLoader);
             schedulerService.leaveJob(jobId, job);
             schedulerService.postJob();
         }
     }
 
 
     @Nullable
     private JobRunnerResponse validate() throws JobRunnerNotRegisteredException {
         JobRunnerResponse response = validateJobDetails();
         if (response == null) {
             response = validateJobRunner();
             if (response == null) {
                 response = validateJobConfig();
             }
         }
         return response;
     }
 
     @Nullable
     private JobRunnerResponse validateJobDetails() {
         if (jobDetails == null) {
             jobDetails = schedulerService.getJobDetails(jobId);
             if (jobDetails == null) {
                 return aborted("No corresponding job details");
             }
         }
 
         if (jobDetails.getRunMode() != schedulerRunMode) {
             return aborted("Inconsistent run mode: expected '" + jobDetails.getRunMode() +
                     "' got: '" + schedulerRunMode + '\'');
         }
         return null;
     }
 
     @Nullable
     private JobRunnerResponse validateJobRunner() throws JobRunnerNotRegisteredException {
         jobRunner = schedulerService.getJobRunner(jobDetails.getJobRunnerKey());
         if (jobRunner == null) {
             // We can't create a JobRunnerResponse for this...
             throw new JobRunnerNotRegisteredException(jobDetails.getJobRunnerKey());
         }
         return null;
     }
 
     @Nullable
     private JobRunnerResponse validateJobConfig() {
         try {
             jobConfig = JobConfig.forJobRunnerKey(jobDetails.getJobRunnerKey())
                     .withRunMode(jobDetails.getRunMode())
                     .withSchedule(jobDetails.getSchedule())
                     .withParameters(jobDetails.getParameters());
             return null;
         } catch (SchedulerRuntimeException sre) {
             return aborted(jobDetails.toString());
         }
     }
 
     private void deleteIfRunOnce() {
         if (jobDetails != null) {
             final IntervalScheduleInfo info = jobDetails.getSchedule().getIntervalScheduleInfo();
             if (info != null && info.getIntervalInMillis() == 0L) {
                 LOG.debug("deleteIfRunOnce: deleting completed job: {}", jobId);
                 schedulerService.unscheduleJob(jobId);
             }
         }
     }
 
     @Override
     public String toString() {
         return "JobLauncher[jobId=" + jobId +
                 ",jobDetails=" + jobDetails +
                 ",response=" + response +
                 ']';
     }
 }