using System;
using System.Collections.Generic;
using System.Configuration;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using DataModel.Models;
using Microsoft.AspNet.SignalR.Client;
using NLog;
using WorkerService.Helpers;
using WorkerService.Models;
using WorkerService.Models.Reporting;
namespace WorkerService.Process
{
///
/// Background work processor
///
public class WorkProcessor : IDisposable
{
///
/// Default wait/sleep interval between processing runs
///
private const int WAIT_INTERVAL = 5000;
///
/// Shared logger instance
///
private static readonly Logger logger = LogManager.GetLogger("WorkProcessor");
///
/// Thread cancellation flag
///
private readonly ManualResetEvent cancel = new ManualResetEvent(false);
///
/// Current work status message
///
public string Status
{
get => _status;
set
{
if (string.Equals(_status, value))
{
return;
}
_status = value;
logger.Info("Status: {0}", _status);
PublishStatus(_status);
}
}
private string _status;
///
/// Publishes status message update to the hub
///
/// Status message to publish
private async void PublishStatus(string status)
{
try
{
string hubHost = ConfigurationManager.AppSettings["HubHost"];
using (HubConnection hubConnection = new HubConnection(hubHost))
{
IHubProxy hubProxy = hubConnection.CreateHubProxy("StatusHub");
await hubConnection.Start();
StatusUpdate update = new StatusUpdate()
{
Message = status,
Timestamp = DateTime.Now
};
await hubProxy.Invoke("SetStatus", update);
}
}
catch (Exception error)
{
logger.Error("PublishStatus: failed to publish status update to hub: {0}.", error.Message);
}
}
///
/// Publishes update for the search to the hub
///
/// Search to publish update for
private async void PublishSearchUpdate(Search search)
{
try
{
string hubHost = ConfigurationManager.AppSettings["HubHost"];
using (HubConnection hubConnection = new HubConnection(hubHost))
{
IHubProxy hubProxy = hubConnection.CreateHubProxy("StatusHub");
await hubConnection.Start();
SearchUpdate searchUpdate = new SearchUpdate(search);
await hubProxy.Invoke("PublishSearchUpdate", searchUpdate);
}
}
catch (Exception error)
{
logger.Error("PublishSearchUpdate: failed to publish search update to hub: {0}.", error.Message);
}
}
///
/// Starts the worker thread
///
public void Start()
{
logger.Info("Background processing thread starting...");
Thread workThread = new Thread(() =>
{
while (true)
{
DoWork();
//Wait to continue
if (cancel.WaitOne(WAIT_INTERVAL))
{
break;
}
}
});
workThread.Start();
}
///
/// Stops the worker thread
///
public void Stop()
{
logger.Info("Background processing thread stopping...");
cancel.Set();
}
///
/// Work processing action
///
private void DoWork()
{
try
{
//Verify all data sources up to date
List pending = UpdateProcessor.GetPendingUpdateTasks();
if (pending.Any())
{
Status = "Updating data cache";
Parallel.ForEach(pending, new ParallelOptions() { MaxDegreeOfParallelism = 8 }, pendingTask => { UpdateProcessor.DoUpdate(pendingTask); });
}
else
{
//Reset any partially completed searches
LotFinderDBExt.ResetPartialSearches();
//Check for queued searches
Search search = LotFinderDBExt.GetNextSearch();
if (search != null)
{
try
{
Status = $"Processing search #{search.ID}";
//Start search
LotFinderDBExt.StartSearch(search);
PublishSearchUpdate(search);
//Do search
SearchModel searchModel = search.ToSearchModel();
LotFinderDBExt.Search(searchModel);
//Record end timestamp
search.EndDT = DateTime.Now;
//Generate output
search.Results = ExcelWriter.Generate(searchModel);
File.WriteAllBytes($"search_{search.ID}.xlsx", search.Results);
//Complete search
LotFinderDBExt.CompleteSearch(search, true);
PublishSearchUpdate(search);
}
catch (Exception error)
{
//Log error and mark search as failed
logger.Error("DoWork: failed to process search: {0}.", error.Message);
search.EndDT = DateTime.Now;
LotFinderDBExt.CompleteSearch(search, false);
PublishSearchUpdate(search);
}
}
}
Status = "Idle";
}
catch (Exception error)
{
//Log but do not forward error
logger.Error("DoWork: work processing run failed: {0}.", error.Message);
}
}
///
/// Stops worker thread
///
public void Dispose()
{
Stop();
}
}
}