package jobmanager.messaging.handler;

import jobmanager.database.MongoAccessor;
import messaging.job.JobMessageFactory;
import model.job.Job;
import model.job.type.AbortJob;
import model.request.PiazzaJobRequest;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.Producer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import util.PiazzaLogger;
import util.UUIDFactory;

import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * Handles the Kafka topic for the requesting of a Job on the "Request-Job"
 * topic. This will relay the "Create-Job" job topic to the appropriate worker
 * components, while additionally adding in the Job metadata to the Jobs table.
 * 
 * @author Patrick.Doody
 *
 */
@Component
public class RequestJobHandler {
	@Autowired
	private PiazzaLogger logger;
	@Autowired
	private MongoAccessor accessor;
	@Autowired
	private UUIDFactory uuidFactory;
	@Autowired
	private AbortJobHandler abortJobHandler;
	@Autowired
	private CreateJobHandler createJobHandler;
	@Value("${SPACE}")
	private String SPACE;

	private Producer<String, String> producer;

	/**
	 * Sets the producer for this Handler. Uses injection from the Job Messager
	 * in order to be efficient in creating only one producer, as producers are
	 * thread-safe.
	 * 
	 * @param producer
	 *            The producer.
	 */
	public void setProducer(Producer<String, String> producer) {
		this.producer = producer;
	}

	/**
	 * Processes a message on the "Request-Job" topic. This will add the Job
	 * metadata into the Jobs table, and then fire the Kafka event to the worker
	 * components to process the Job. 
	 * 
	 * @param consumerRecord
	 *            The Job request message.
	 */
	@Async
	public void process(ConsumerRecord<String, String> consumerRecord) {
		try {
			// Deserialize the message
			ObjectMapper mapper = new ObjectMapper();
			PiazzaJobRequest jobRequest = mapper.readValue(consumerRecord.value(), PiazzaJobRequest.class);
			String jobId = consumerRecord.key();
			process(jobRequest, jobId);
		} catch (Exception exception) {
			logger.log(String.format("Error Processing Request-Job Topic %s with key %s", consumerRecord.topic(),
					consumerRecord.key()), PiazzaLogger.ERROR);
			exception.printStackTrace();
		}
	}

	/**
	 * Processes a new Piazza Job Request. This will add the Job metadata into
	 * the Jobs table, and then fire the Kafka event to the worker components to
	 * process the Job. 
	 * 
	 * @param jobRequest
	 *            The Job Request
	 * @param jobId
	 *            the Job ID
	 */
	@Async
	public void process(PiazzaJobRequest jobRequest, String jobId) {
		try {
			Job job = new Job(jobRequest, jobId);
			// If the job was submitted internally, the submitter
			// wouldn't give it an ID. Assign a random ID here. (If
			// submitted via the Gateway, the Gateway will assign
			// the ID)
			if (job.getJobId().isEmpty()) {
				job.setJobId(uuidFactory.getUUID());
			}

			// The Legacy API needs the Dispatcher to handle the
			// AbortJob type. When the Legacy API becomes
			// deprecated, then this AbortJob handler will be
			// removed.
			if (job.jobType instanceof AbortJob) {
				abortJobHandler.process(jobRequest);
			} else {
				// Commit the Job metadata to the Jobs table
				createJobHandler.process(job);
				// Send the content of the actual Job under the
				// topic name of the Job type for all workers to
				// listen to.
				producer.send(JobMessageFactory.getWorkerJobCreateMessage(job, SPACE)).get();
			}
			logger.log(String.format("Relayed Job ID %s for Type %s", job.getJobId(), job.getJobType().getClass().getSimpleName()),
					PiazzaLogger.INFO);
		} catch (Exception exception) {
			logger.log(String.format("Error Processing Request-Job."), PiazzaLogger.ERROR);
			exception.printStackTrace();
		}
	}
}

@@ -140,40 +140,8 @@
 ob;%0A
-import model.job.type.AbortJob;%0A
 impo
@@ -175,16 +175,16 @@
 equest;%0A
+
 %0Aimport 
@@ -1780,33 +1780,32 @@
 process the Job.
- 
 %0A%09 * %0A%09 * @param
@@ -2603,17 +2603,16 @@
 the Job.
- 
 %0A%09 * %0A%09 
@@ -3105,278 +3105,8 @@
 %09%09%7D%0A
-%0A%09%09%09// The Legacy API needs the Dispatcher to handle the%0A%09%09%09// AbortJob type. When the Legacy API becomes%0A%09%09%09// deprecated, then this AbortJob handler will be%0A%09%09%09// removed.%0A%09%09%09if (job.jobType instanceof AbortJob) %7B%0A%09%09%09%09abortJobHandler.process(jobRequest);%0A%09%09%09%7D else %7B%0A%09
 %09%09%09/
@@ -3152,17 +3152,16 @@
 able%0A%09%09%09
-%09
 createJo
@@ -3183,17 +3183,16 @@
 s(job);%0A
-%09
 %09%09%09// Se
@@ -3237,17 +3237,16 @@
  the%0A%09%09%09
-%09
 // topic
@@ -3286,17 +3286,16 @@
 kers to%0A
-%09
 %09%09%09// li
@@ -3306,17 +3306,16 @@
  to.%0A%09%09%09
-%09
 producer
@@ -3384,21 +3384,16 @@
 .get();%0A
-%09%09%09%7D%0A
 %09%09%09logge
@@ -3390,32 +3390,38 @@
 ;%0A%09%09%09logger.log(
+%0A%09%09%09%09%09
 String.format(%22R
@@ -3494,16 +3494,24 @@
 tClass()
+%0A%09%09%09%09%09%09%09
 .getSimp
@@ -3520,22 +3520,17 @@
 Name()),
-%0A%09%09%09%09%09
+ 
 PiazzaLo

Remove dispatcher reference