package gateway.controller;

import gateway.auth.AuthConnector;

import java.util.UUID;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import messaging.job.JobMessageFactory;
import messaging.job.KafkaClientFactory;
import messaging.job.async.JobMessageSyncUtility;
import model.job.type.GetJob;
import model.request.PiazzaJobRequest;
import model.response.ErrorResponse;
import model.response.PiazzaResponse;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;
import org.springframework.web.multipart.MultipartFile;

import com.fasterxml.jackson.core.JsonProcessingException;

@RestController
public class GatewayController {
	/**
	 * The Kafka Producer that will send messages from this controller to the
	 * Dispatcher. Initialized upon Controller startup.
	 */
	private Producer<String, String> producer;
	@Value("${kafka.host}")
	private String KAFKA_HOST;
	@Value("${kafka.port}")
	private String KAFKA_PORT;
	@Value("${kafka.group}")
	private String KAFKA_GROUP;
	@Value("${dispatcher.host}")
	private String DISPATCHER_HOST;
	@Value("${dispatcher.port}")
	private String DISPATCHER_PORT;

	/**
	 * Initializing the Kafka Producer on Controller startup.
	 */
	@PostConstruct
	public void init() {
		producer = KafkaClientFactory.getProducer(KAFKA_HOST, KAFKA_PORT);
	}

	@PreDestroy
	public void cleanup() {
		producer.close();
	}

	/**
	 * Executes a Piazza Job
	 * 
	 * @param json
	 *            The JSON Payload
	 * @return Response object.
	 */
	@RequestMapping(value = "/job", method = RequestMethod.POST)
	public PiazzaResponse job(@RequestParam(required = true) String body,
			@RequestParam(required = false) final MultipartFile file) {

		// Deserialize the incoming JSON to Request Model objects
		PiazzaJobRequest request;
		try {
			request = JobMessageFactory.parseRequestJson(body);
		} catch (Exception exception) {
			return new ErrorResponse(null, "Error Parsing JSON: " + exception.getMessage(), "Gateway");
		}

		// Authenticate and Authorize the request
		try {
			AuthConnector.verifyAuth(request);
		} catch (SecurityException securityEx) {
			return new ErrorResponse(null, "Authentication Error", "Gateway");
		}

		// Determine if this Job is processed via synchronous REST, or via Kafka
		// message queues.
		if (request.jobType instanceof GetJob) {
			// Ensure the Job ID is populated.
			GetJob getJob = (GetJob) request.jobType;
			if (getJob.getJobId() == null || getJob.getJobId().isEmpty()) {
				return new ErrorResponse(null, "Missing JobID Property.", "Gateway");
			}
			// REST GET request to Dispatcher to fetch the status of the Job ID.
			RestTemplate restTemplate = new RestTemplate();
			try {
				PiazzaResponse dispatcherResponse = restTemplate.getForObject(
						String.format("http://%s:%s/job/%s", DISPATCHER_HOST, DISPATCHER_PORT, getJob.getJobId()),
						PiazzaResponse.class);
				return dispatcherResponse;
			} catch (RestClientException exception) {
				return new ErrorResponse(null, "Error connecting to Dispatcher service: " + exception.getMessage(),
						"Gateway");
			}
		} else {
			// Create a GUID for this new Job.
			String jobId = UUID.randomUUID().toString();
			// Create the Kafka Message for an incoming Job to be created.
			final ProducerRecord<String, String> message;
			try {
				message = JobMessageFactory.getRequestJobMessage(request, jobId);
			} catch (JsonProcessingException exception) {
				return new ErrorResponse(jobId, "Error Creating Message for Job", "Gateway");
			}

			System.out.println("Requesting Job topic " + message.topic() + " with key " + message.key());

			// Fire off a Kafka Message and then wait for a response from the
			// Job Manager that the Job has been indexed.
			JobMessageSyncUtility syncUtility = new JobMessageSyncUtility(KAFKA_HOST, KAFKA_PORT, KAFKA_GROUP, message,
					10);
			Future<ConsumerRecord<String, String>> jobStatus = Executors.newSingleThreadExecutor().submit(syncUtility);

			// Wait for job future to be fulfilled. This signifies that the Job
			// has been received by the Job Manager.
			while (!jobStatus.isDone()) {
				try {
					Thread.sleep(1000);
				} catch (InterruptedException exception) {
					exception.printStackTrace();
					return new ErrorResponse(jobId, "Thread Management Error", "Gateway");
				}
			}

			// Ensure no Exceptions were produced by the ASync callable.
			try {
				jobStatus.get();
			} catch (Exception exception) {
				return new ErrorResponse(
						jobId,
						"The Gateway did not receive a response from the Job Manager; the request may not have succeeded. Please query the Job ID to verify if the Job was successfully submitted.",
						"Gateway");
			}

			// Respond immediately with the new Job GUID
			return new PiazzaResponse(jobId);
		}
	}
}
@@ -84,131 +84,8 @@
 UID;
-%0Aimport java.util.concurrent.ExecutionException;%0Aimport java.util.concurrent.Executors;%0Aimport java.util.concurrent.Future;
 %0A%0Aim
@@ -243,58 +243,8 @@
 ry;%0A
-import messaging.job.async.JobMessageSyncUtility;%0A
 impo
@@ -388,65 +388,8 @@
 e;%0A%0A
-import org.apache.kafka.clients.consumer.ConsumerRecord;%0A
 impo
@@ -4070,16 +4070,20 @@
 t for a 
+ack 
 response
@@ -4091,733 +4091,67 @@
 from
- the
 %0A%09%09%09// 
-Job Manager that the Job has been indexed.%0A%09%09%09JobMessageSyncUtility syncUtility = new JobMessageSyncUtility(KAFKA_HOST, KAFKA_PORT, KAFKA_GROUP, message,%0A%09%09%09%09%0910);%0A%09%09%09Future%3CConsumerRecord%3CString, String%3E%3E jobStatus = Executors.newSingleThreadExecutor().submit(syncUtility);%0A%0A%09%09%09// Wait for job future to be fulfilled. This signifies that the Job%0A%09%09%09// has been received by the Job Manag
+the kafka brok
 er
-.
 %0A%09%09%09
-while (!jobStatus.isDone()) %7B%0A%09%09%09%09try %7B%0A%09%09%09%09%09Thread.sleep(1000);%0A%09%09%09%09%7D catch (InterruptedException exception) %7B%0A%09%09%09%09%09exception.printStackTrace();%0A%09%09%09%09%09return new ErrorResponse(jobId, %22Thread Management Error%22, %22Gateway%22);%0A%09%09%09%09%7D%0A%09%09%09%7D%0A%0A%09%09%09// Ensure no Exceptions were produced by the ASync callable.%0A%09%09%09try %7B%0A%09%09%09%09jobStatus
+try %7B%0A%09%09%09%09producer.send(message)
 .get
@@ -4287,132 +4287,65 @@
 rom 
-the Job Manager; the request may not have succeeded. Please query the Job ID to verify if the Job was successfully submitted
+Kafka; the request could not be forwarded along to Piazza
 .%22,%0A

Revert async listener API