package ch.hsr.camel.jobs.trigger;

import org.apache.camel.Exchange;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.Route;
import org.apache.camel.component.seda.SedaEndpoint;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.UnexpectedRollbackException;
import org.springframework.util.StopWatch;

import ch.hsr.camel.jobs.trigger.rest.MetricHelper;
import ch.hsr.camel.jobs.trigger.rest.api.JobResult;
import ch.hsr.camel.jobs.trigger.rest.api.JobResult.Status;

@Component("jobManager")
public class JobManager {

	@Autowired
	private ProducerTemplate template;

	@Autowired
	MetricHelper metricHelper;

	private Logger logger = LoggerFactory.getLogger(JobManager.class);

	public JobManager() {
	}

	/**
	 * start a job
	 * 
	 * @param jobName
	 * @throws IllegalArgumentException
	 *             when the job does not exist.
	 * @throws IllegalArgumentException
	 *             when a job is started that cannot be started by the JobManager.
	 * @throws IllegalStateException
	 *             when there is a problem with job naming.
	 */
	public JobResult startJob(String jobName) {
		String executionId;
		Status status = JobResult.Status.SUCCESSFUL;
		long countSuccessfulBefore = metricHelper.getSuccessfulRecords(jobName);
		long countFailedBefore = metricHelper.getFailedRecords(jobName);
		long countIgnoredBefore = metricHelper.getIgnoredRecords(jobName);
		long countRejectedBefore = metricHelper.getRejectedRecords(jobName);

		StopWatch sw = new StopWatch();
		String endpointUri = "vm:trigger-" + jobName;
		SedaEndpoint endpoint = (SedaEndpoint) template.getCamelContext().hasEndpoint(endpointUri);
		// TODO instanceof check!

		Route route = template.getCamelContext().getRoute(jobName);

		if (route == null) {
			throw new IllegalArgumentException("Job " + jobName + " not found.");
		} else if (route.getEndpoint().getEndpointUri().startsWith("file://")) {
			throw new IllegalArgumentException("Job " + jobName
					+ " is based on a file poller and cannot be started manually.");
		} else if (endpoint != null) {
			// cannot set timeout in endpointUri as it won't be found by hasEndpoint
			endpoint.setTimeout(0);
			sw.start();
			logger.info("Starting job {}.", jobName);
			Exchange result = template.request(endpoint, null);
			if (result.getException() != null) {
				logger.error("Job {} failed.", jobName, result.getException());
				if (result.getException() instanceof UnexpectedRollbackException) {
					status = JobResult.Status.ROLLED_BACK;
				} else {
					status = JobResult.Status.FAILED;
				}
			}
			executionId = getExecutionId(result);
			sw.stop();
		} else {
			throw new IllegalStateException("Jobs are not named according to the convention. Endpointuri: "
					+ endpointUri);
		}

		JobResult jobResult = new JobResult(jobName, status, executionId);
		long successfulRecords = metricHelper.getSuccessfulRecords(jobName) - countSuccessfulBefore;
		long ignoredRecords = metricHelper.getIgnoredRecords(jobName) - countIgnoredBefore;
		long failedRecords = metricHelper.getFailedRecords(jobName) - countFailedBefore;
		long rejectedRecords = metricHelper.getRejectedRecords(jobName) - countRejectedBefore;
		jobResult.setSuccessfulRecords(successfulRecords);
		jobResult.setFailedRecords(failedRecords);
		jobResult.setIgnoredRecords(ignoredRecords);
		jobResult.setRejectedRecords(rejectedRecords);

		logger.info("Job {} finished in {}ms. {}", jobName, sw.getLastTaskTimeMillis(), jobResult.getDetails());

		return jobResult;
	}

	private String getExecutionId(Exchange result) {
		if (result.getOut() != null) {
			Object executionIdHeader = result.getOut().getHeader("ExecutionId");
			if (executionIdHeader != null) {
				return executionIdHeader.toString();
			}
		}
		return "";
	}
}

@@ -63,16 +63,83 @@
 change;%0A
+import org.apache.camel.Header;%0Aimport org.apache.camel.Processor;%0A
 import o
@@ -168,24 +168,24 @@
 erTemplate;%0A
-
 import org.a
@@ -941,16 +941,242 @@
 ) %7B%0A%09%7D%0A%0A
+%09public JobResult startJob(String jobName) %7B%0A%09%09return startJob(jobName, null);%0A%09%7D%0A%0A%09public JobResult startJob(@Header(%22jobName%22) String jobName, Exchange exchange) %7B%0A%09%09return startJob(jobName, exchange.getIn().getBody());%0A%09%7D%0A%0A
 %09/**%0A%09 *
@@ -1544,24 +1544,43 @@
 ring jobName
+, final Object data
 ) %7B%0A%09%09String
@@ -2732,19 +2732,156 @@
 point, n
-ull
+ew Processor() %7B%0A%09%09%09%09@Override%0A%09%09%09%09public void process(Exchange exchange) throws Exception %7B%0A%09%09%09%09%09exchange.getIn().setBody(data);%0A%09%09%09%09%7D%0A%09%09%09%7D
 );%0A%09%09%09if

handle all job executions by jobManager