qdr.fnd/qdr.fnd.core.pqueue/Process/PsProcessProviderBase.cs

using qdr.fnd.core.pqueue.Enums;
using qdr.fnd.core.pqueue.Exceptions;
using qdr.fnd.core.pqueue.fnd.candidates;
using Quadarax.Foundation.Core.Data;
using Quadarax.Foundation.Core.Logging;
using Quadarax.Foundation.Core.Object;
using Quadarax.Foundation.Core.Value.Extensions;

namespace qdr.fnd.core.pqueue.Process
{
    public delegate void ProcessDoneCallback(string itemId, PsStatusEnum newStatus, string message);
    public delegate void ProcessStdOutputCallback(string itemId, string message);

    public abstract class PsProcessProviderBase : DisposableObject, IPsProcessProvider
    {

        #region *** Constants ***
        /// <summary>
        /// (Output) Argument name for the result code. Type is <see cref="long"/>
        /// </summary>
        public const string CS_ARG_RESULT_CODE = "result-code";
        /// <summary>
        /// (Output) Argument name for the result test. Type is <see cref="string"/>
        /// </summary>
        public const string CS_ARG_RESULT_TEXT = "result-text";
        /// <summary>
        /// (Input) Argument name for process timeout. Type is <see cref="TimeSpan"/>
        /// </summary>
        public const string CS_ARG_PROCESS_TIMEOUT = "process-timeout";
        #endregion

        #region *** Properties ***
        protected ILog Log { get; private set; }
        protected ProcessDoneCallback ProcessDoneCallback { get; private set; }
        protected ProcessStdOutputCallback ProcessStdOutputCallback { get; private set; }
        /// <inheritdoc />
        public IDictionary<string, TypedArgument> Arguments { get; private set; }
        public bool IsRunning { get; private set; }
        /// <inheritdoc />
        public string ItemId { get; private set; }
        /// <inheritdoc />
        public DateTime? StartedTimestamp { get; private set; }
        /// <inheritdoc />
        public DateTime? FinishedTimestamp { get; private set; }

        #endregion

        #region *** Private fields ***
        private CancellationTokenSource? _cancelationToken;
        #endregion

        #region *** Constructors ***
        public PsProcessProviderBase(ProcessDoneCallback processDoneCallback, ProcessStdOutputCallback processStdOutputCallback, string itemId, TypedArgument[] arguments, ILogger logger)
        {
            if (string.IsNullOrEmpty(itemId)) throw new ArgumentNullException(nameof(itemId));
            ProcessDoneCallback = processDoneCallback ?? throw new ArgumentNullException(nameof(processDoneCallback));
            ProcessStdOutputCallback = processStdOutputCallback ?? throw new ArgumentNullException(nameof(processStdOutputCallback));
            ItemId = itemId;
            Log = logger.GetLogger(GetType());
            Arguments = arguments.ToDictionary(x => x.Name, x => x);
            _cancelationToken = new CancellationTokenSource();
        }
        #endregion

        #region *** Public Operations ***

        /// <inheritdoc />
        public void Validate()
        {
            OnValidate();
        }
        /// <inheritdoc />
        public void Start()
        {
            if (IsRunning)
                throw new PsProcessProviderException(PsProcessProviderException.ErrorCodes.ProcessProviderIsRunning, GetType().Name, ItemId);

            Validate();
            Log.Log(LogSeverityEnum.Info, $"Provider for ItemId='{ItemId}' starting.");
            IsRunning = true;
            StartedTimestamp = DateTime.Now;
            var timeout = GetArgumentValue<TimeSpan>(CS_ARG_PROCESS_TIMEOUT);


            Task.Run(async () =>
            {

                try
                {
                    await TimeoutUtils.ExecuteWithTimeoutAsync(async () =>
                    {

                        await OnProcess().ContinueWith(task =>
                        {
                            if (task.IsFaulted)
                            {
                                var e = task.Exception!;
                                OnProcessError(e).ContinueWith(taskInner =>
                                    {
                                        if (taskInner.IsFaulted)
                                            throw taskInner.Exception!;
                                        ProcessDoneFailed(e.Message, e);
                                    });
                            }
                            else
                            {
                                ProcessDoneOk("Succesfuly done.");
                            }
                        });
                    }, timeout);
                }
                catch (Exception e)
                {
                    await OnProcessError(e).ContinueWith(task =>
                    {
                        if (task.IsFaulted)
                            throw task.Exception!;
                        ProcessDoneFailed(e.Message, e);
                    });
                }
            });
        }

        /// <inheritdoc />
        public void Stop()
        {
            if (!IsRunning)
                throw new PsProcessProviderException(PsProcessProviderException.ErrorCodes.ProcessProviderNotRunning, GetType().Name, ItemId);

            _cancelationToken?.Cancel();
            ProcessDoneFailed("Process was forced stopped.");
            Log.Log(LogSeverityEnum.Info, $"Provider for ItemId='{ItemId}' was forced stopped.");
        }

        /// <summary>
        /// Generates all arguments collection
        /// </summary>
        /// <returns>Full scoped arguments collection</returns>
        public static IEnumerable<TypedArgument> CreateArguments()
        {
            var list = new List<TypedArgument>
            {
                new TypedArgument(CS_ARG_RESULT_CODE, typeof(long), true),
                new TypedArgument(CS_ARG_RESULT_TEXT, typeof(string), true),
                new TypedArgument(CS_ARG_PROCESS_TIMEOUT, typeof(TimeSpan), false)
            };
            return list;
        }

        /// <inheritdoc />
        public void WaitToDone()
        {
            while (IsRunning)
            {
                Task.Delay(100).Wait();
            };
            Log.Log(LogSeverityEnum.Debug, $"Provider for ItemId='{ItemId}' detect that process finished.");
        }
        #endregion


        #region *** Protected Operations ***
        protected TValue? GetArgumentValue<TValue>(string name)
        {
            if (string.IsNullOrEmpty(name))
                throw new ArgumentNullException(nameof(name));

            if (Arguments.TryGetValue(name, out var arg))
                return arg.GetTypedValue<TValue>();
            return default;
        }
        protected bool ContainsArgument(string name)
        {
            if (string.IsNullOrEmpty(name))
                throw new ArgumentNullException(nameof(name));

            return Arguments.ContainsKey(name);
        }

        protected virtual void OnValidate()
        {
            if (!ContainsArgument(CS_ARG_RESULT_CODE) || !Arguments[CS_ARG_RESULT_CODE].IsOutput)
                throw new PsProcessProviderException(PsProcessProviderException.ErrorCodes.ProcessProviderNotDefinedArgument, GetType().Name, ItemId, CS_ARG_RESULT_CODE);
            if (!ContainsArgument(CS_ARG_RESULT_TEXT) || !Arguments[CS_ARG_RESULT_TEXT].IsOutput)
                throw new PsProcessProviderException(PsProcessProviderException.ErrorCodes.ProcessProviderNotDefinedArgument, GetType().Name, ItemId, CS_ARG_RESULT_TEXT);
            if (!ContainsArgument(CS_ARG_PROCESS_TIMEOUT) || Arguments[CS_ARG_PROCESS_TIMEOUT].IsOutput)
                throw new PsProcessProviderException(PsProcessProviderException.ErrorCodes.ProcessProviderNotDefinedArgument, GetType().Name, ItemId, CS_ARG_PROCESS_TIMEOUT);
        }
#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
        protected virtual async Task OnProcessError(Exception exception)
#pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously
        {
            if (exception == null)
                throw new ArgumentNullException(nameof(exception));

            if (ProcessStdOutputCallback != null)
                ProcessStdOutputCallback(ItemId, exception.Message);

            if (ContainsArgument(CS_ARG_RESULT_TEXT))
                Arguments[CS_ARG_RESULT_TEXT].Value = exception.Message;
        }

#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
        protected virtual async Task OnProcess()
#pragma warning restore CS1998 // Async method lacks 'await' operators and will run synchronously
        {
            //nothing to do
        }

        /// <inheritdoc />
        protected override void OnDisposing()
        {
            _cancelationToken?.Dispose();
            ProcessDoneCallback = null!;
            ProcessStdOutputCallback = null!;
            _cancelationToken = null;
        }


        private void ProcessDoneFailed(string message, Exception? e = null)
        {
            Log.Log(LogSeverityEnum.Error, $"Provider running failed. Duration {(DateTime.Now - StartedTimestamp.GetValueOrDefault()).ToReadableString()}.", e);
            if (ProcessDoneCallback != null)
                ProcessDoneCallback(ItemId, PsStatusEnum.DoneFailed, message);
            IsRunning = false;
            FinishedTimestamp = DateTime.Now;
        }
        private void ProcessDoneOk(string message)
        {
            Log.Log(LogSeverityEnum.Info, $"Provider for ItemId='{ItemId}' done. Duration {(DateTime.Now - StartedTimestamp.GetValueOrDefault()).ToReadableString()}.");
            if (ProcessDoneCallback != null)
                ProcessDoneCallback(ItemId, PsStatusEnum.DoneOk, message);
            IsRunning = false;
            FinishedTimestamp = DateTime.Now;
        }


        #endregion

    }

}