

package org.apache.samza.job.local

import java.lang.Thread.UncaughtExceptionHandler

import org.apache.samza.util.Logging
import org.apache.samza.job.StreamJob
import org.apache.samza.job.ApplicationStatus
import org.apache.samza.job.ApplicationStatus.New
import org.apache.samza.job.ApplicationStatus.Running
import org.apache.samza.job.ApplicationStatus.SuccessfulFinish
import org.apache.samza.job.ApplicationStatus.UnsuccessfulFinish

class ThreadJob(runnable: Runnable) extends StreamJob with Logging {
  @volatile var jobStatus: Option[ApplicationStatus] = None
  var thread: Thread = null

  def submit: StreamJob = {
    jobStatus = Some(New)

    // create a non-daemon thread to make job runner block until the job finishes.
    // without this, the proc dies when job runner ends.
    thread = new Thread {
      override def run {
        try {
          runnable.run
          jobStatus = Some(SuccessfulFinish)
        } catch {
          case e: Throwable => {
            error("Failing job with exception.", e)
            jobStatus = Some(UnsuccessfulFinish)
            throw e
          }
        }
      }
    }
    thread.setName("ThreadJob")
    thread.start
    jobStatus = Some(Running)

    ThreadJob.this
  }

  def kill: StreamJob = {
    thread.interrupt
    ThreadJob.this
  }

  def waitForFinish(timeoutMs: Long) = {
    thread.join(timeoutMs)
    jobStatus.getOrElse(null)
  }

  def waitForStatus(status: ApplicationStatus, timeoutMs: Long) = {
    val start = System.currentTimeMillis

    while (System.currentTimeMillis - start < timeoutMs && !status.equals(jobStatus.getOrElse(null))) {
      Thread.sleep(500)
    }

    jobStatus.getOrElse(null)
  }

  def getStatus = jobStatus.getOrElse(null)
}
