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 *** /// /// (Output) Argument name for the result code. Type is /// public const string CS_ARG_RESULT_CODE = "result-code"; /// /// (Output) Argument name for the result test. Type is /// public const string CS_ARG_RESULT_TEXT = "result-text"; /// /// (Input) Argument name for process timeout. Type is /// 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; } /// public IDictionary Arguments { get; private set; } public bool IsRunning { get; private set; } /// public string ItemId { get; private set; } /// public DateTime? StartedTimestamp { get; private set; } /// public DateTime? FinishedTimestamp { get; private set; } #endregion #region *** Private fields *** private CancellationTokenSource? _cancelationToken; #endregion #region *** Constructors *** public PsProcessProviderBase(Callbacks callbacks, string itemId, TypedArgument[] arguments, ILogger logger) { if (string.IsNullOrEmpty(itemId)) throw new ArgumentNullException(nameof(itemId)); if (callbacks==null) throw new ArgumentNullException(nameof(callbacks)); ProcessDoneCallback = callbacks.ProcessDoneCallback ?? throw new ArgumentNullException("callbacks.ProcessDoneCallback"); ProcessStdOutputCallback = callbacks.ProcessStdOutputCallback ?? throw new ArgumentNullException("callbacks.ProcessStdOutputCallback"); ItemId = itemId; Log = logger.GetLogger(GetType()); Arguments = arguments.ToDictionary(x => x.Name, x => x); _cancelationToken = new CancellationTokenSource(); } public PsProcessProviderBase(ConstructorArguments constructorArguments) : this(constructorArguments.Callbacks, constructorArguments.ItemId, constructorArguments.Arguments, constructorArguments.Logger) { } #endregion #region *** Public Operations *** /// public void Validate() { OnValidate(); } /// 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(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); }); } }); } /// 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."); } /// /// Generates all arguments collection /// /// Full scoped arguments collection public static IEnumerable CreateArguments() { var list = new List { 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; } /// 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(string name) { if (string.IsNullOrEmpty(name)) throw new ArgumentNullException(nameof(name)); if (Arguments.TryGetValue(name, out var arg)) return arg.GetTypedValue(); 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 } /// 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 #region *** Nested Classes *** public class Callbacks { public ProcessDoneCallback ProcessDoneCallback { get; private set; } public ProcessStdOutputCallback ProcessStdOutputCallback { get; private set; } public Callbacks(ProcessDoneCallback processDoneCallback, ProcessStdOutputCallback processStdOutputCallback) { ProcessDoneCallback = processDoneCallback; ProcessStdOutputCallback = processStdOutputCallback; } } public class ConstructorArguments { public Callbacks Callbacks { get; private set; } public string ItemId { get; private set; } public TypedArgument[] Arguments { get; private set; } public ILogger Logger { get; private set; } public ConstructorArguments(Callbacks callbacks, string itemId, TypedArgument[] arguments, ILogger logger) { Callbacks = callbacks; ItemId = itemId; Arguments = arguments; Logger = logger; } } #endregion } }