jdescopingtool/OLD/WorkerService/Process/WorkProcessor.cs

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
{
    /// <summary>
    /// Background work processor
    /// </summary>
    public class WorkProcessor : IDisposable
    {
        /// <summary>
        /// Default wait/sleep interval between processing runs
        /// </summary>
        private const int WAIT_INTERVAL = 5000;

        /// <summary>
        /// Shared logger instance
        /// </summary>
        private static readonly Logger logger = LogManager.GetLogger("WorkProcessor");

        /// <summary>
        /// Thread cancellation flag
        /// </summary>
        private readonly ManualResetEvent cancel = new ManualResetEvent(false);

        /// <summary>
        /// Current work status message
        /// </summary>
        public string Status
        {
            get => _status;
            set
            {
                if (string.Equals(_status, value))
                {
                    return;
                }

                _status = value;
                logger.Info("Status: {0}", _status);
                PublishStatus(_status);
            }
        }
        private string _status;

        /// <summary>
        /// Publishes status message update to the hub
        /// </summary>
        /// <param name="status">Status message to publish</param>
        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);
            }
        }

        /// <summary>
        /// Publishes update for the search to the hub
        /// </summary>
        /// <param name="search">Search to publish update for</param>
        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);
            }
        }

        /// <summary>
        /// Starts the worker thread
        /// </summary>
        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();
        }

        /// <summary>
        /// Stops the worker thread
        /// </summary>
        public void Stop()
        {
            logger.Info("Background processing thread stopping...");

            cancel.Set();
        }

        /// <summary>
        /// Work processing action
        /// </summary>
        private void DoWork()
        {
            try
            {
                //Verify all data sources up to date
                List<DataUpdateTask> 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);
            }
        }

        /// <summary>
        /// Stops worker thread
        /// </summary>
        public void Dispose()
        {
            Stop();
        }
    }
}