LeafWeb/WebCms/Services/PiscalQueue/PiscalQueueManager.cs

using System;
using System.Linq;
using System.Threading;
using LeafWeb.Core.Entities;
using LeafWeb.Core.Remote;

namespace LeafWeb.WebCms.Services.PiscalQueue
{
	public class PiscalQueueManager : PiscalQueueBase
	{
		private static readonly object ProcessQueueLock = new object();

		public void ProcessQueue()
		{
			// prevent multiple entry into processing the queue
			if (Monitor.TryEnter(ProcessQueueLock))
			{
				Logger.DebugFormat("ProcessQueue entered");

				try
				{
					UpdateRunning();

					StartNextPending();

					// TODO: handle starting and finishing
				}
				finally
				{
					Logger.DebugFormat("ProcessQueue exit");

					Monitor.Exit(ProcessQueueLock);
				}
			}
			else
			{
				Logger.DebugFormat("ProcessQueue locked, queue already processing");
			}
		}

		// TODO: clear any stalled processes
		private void ClearStalled()
		{
			var leafInputs = 
				DataService.GetLeafInputs(
					LeafInputStatusType.Starting, 
					LeafInputStatusType.Finishing
					)
					.Where(li => 
						li.StatusHistory.OrderBy(sh => sh.DateTime).First().DateTime
							> DateTime.Now.Subtract(TimeSpan.FromHours(1)))
					.ToList();
		}

		private void StartNextPending()
		{
			var runningLeafInputs = 
				DataService.GetLeafInputs(
					LeafInputStatusType.Starting,
					LeafInputStatusType.Running,
					LeafInputStatusType.Finishing 
					).ToList();

			if (runningLeafInputs.Any())
			{
				Logger.DebugFormat("Leaf input(s) currently running");
				return;
			}

			var pendingInput = 
				DataService
					.GetLeafInputs(LeafInputStatusType.Pending)
					.OrderBy(l => l.StatusHistory.Min(sh => sh.DateTime))
					.FirstOrDefault();

			if (pendingInput == null)
			{
				Logger.DebugFormat("No pending leaf input");
				return;
			}

			var pendingInputId = pendingInput.Id;
			Logger.InfoFormat("LeafInput: {0}, Starting", pendingInputId);
			try
			{
				DataService.SetLeafInputStatus(pendingInput, LeafInputStatusType.Starting, "Copying input to Piscal",
					$"File count: {pendingInput.InputFiles.Count}");
				BackgroundJobEnqueueRetry<StartPending>(c => c.DoWork(pendingInputId));
			}
			catch (Exception ex)
			{
				var errorMessage = FormatException(ex);
				Logger.Error(errorMessage);
				DataService.SetLeafInputStatus(pendingInput, LeafInputStatusType.Exception, "Exception while StartPending", errorMessage);
			}
		}

		private void UpdateRunning()
		{
			var running = DataService.GetLeafInputs(LeafInputStatusType.Running).ToList();
			foreach (var leafInput in running)
			{
				try
				{
					var status = PiscalService.GetStatus(leafInput);
					var leafInputId = leafInput.Id;
					switch (status)
					{
						case PiscalStatus.NotStarted:
							// if it's not started - this is unusual state
							var pendingToRetry = "Piscal reporting Not Started, setting to Pending to retry";
							PiscalWarningHandler($"LeafInput: {leafInput.Id}, {pendingToRetry}", leafInput);
							DataService.SetLeafInputStatus(leafInput, LeafInputStatusType.Pending, details: pendingToRetry);
							break;

						case PiscalStatus.Running:
							Logger.DebugFormat("LeafInput: {0}, Piscal Running", leafInput.Id);
							// continue running
							break;

						case PiscalStatus.Complete:
							DataService.SetLeafInputStatus(leafInput, LeafInputStatusType.Finishing, "Copying LeafOutput from Piscal");
							BackgroundJobEnqueueRetry<FinishComplete>(s => s.DoWork(leafInputId));
							
							break;
					}
				}
				catch (PiscalClientException ex)
				{
					PiscalExceptionHandler(ex, leafInput);
				}
				catch (Exception ex)
				{
					var errorMessage = FormatException(ex);
					Logger.Error(errorMessage);
				}
			}
		}
	}
}