TaskItem.cs 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364
  1. using System.Text;
  2. using System.Threading;
  3. using Quadarax.Foundation.Core.Business.Processor.Enums;
  4. using Quadarax.Foundation.Core.Business.Processor.Exceptons;
  5. using Quadarax.Foundation.Core.Business.Processor.Queue;
  6. using Quadarax.Foundation.Core.Data;
  7. using Quadarax.Foundation.Core.Logging;
  8. using Quadarax.Foundation.Core.Object;
  9. using Quadarax.Foundation.Core.Value.Extensions;
  10. namespace Quadarax.Foundation.Core.Business.Processor.Task
  11. {
  12. internal class TaskItem : ITaskItem
  13. {
  14. #region *** Properties ***
  15. /// <summary>
  16. /// <inheritdoc cref="ITaskItem.Id"/>
  17. /// </summary>
  18. public long Id { get; internal set; }
  19. /// <summary>
  20. /// <inheritdoc cref="ITaskItem.QueueType"/>
  21. /// </summary>
  22. public ProcessQueueTypeEnum QueueType { get;private set; }
  23. /// <summary>
  24. /// <inheritdoc cref="ITaskItem.Created"/>
  25. /// </summary>
  26. public DateTime Created { get; private set;}
  27. /// <summary>
  28. /// <inheritdoc cref="ITaskItem.Changed"/>
  29. /// </summary>
  30. public DateTime? Changed { get; private set;}
  31. /// <summary>
  32. /// <inheritdoc cref="ITaskItem.AffinityToken"/>
  33. /// </summary>
  34. public IAffinityToken? AffinityToken { get; private set;}
  35. /// <summary>
  36. /// <inheritdoc cref="ITaskItem.Reference"/>
  37. /// </summary>
  38. public string Reference { get; private set;}
  39. /// <summary>
  40. /// <inheritdoc cref="ITaskItem.Status"/>
  41. /// </summary>
  42. public ProcessStatusEnum Status { get; private set; }
  43. /// <summary>
  44. /// <inheritdoc cref="ITaskItem.ProviderClass"/>
  45. /// </summary>
  46. public string ProviderClass { get; private set;}
  47. /// <summary>
  48. /// <inheritdoc cref="ITaskItem.Priority"/>
  49. /// </summary>
  50. public byte Priority { get; private set;}
  51. /// <summary>
  52. /// <inheritdoc cref="ITaskItem.ProcessingValidFrom"/>
  53. /// </summary>
  54. public DateTime ProcessingValidFrom { get;private set; }
  55. /// <summary>
  56. /// <inheritdoc cref="ITaskItem.ProcessingValidTo"/>
  57. /// </summary>
  58. public DateTime? ProcessingValidTo { get; private set;}
  59. /// <summary>
  60. /// <inheritdoc cref="ITaskItem.AttemptsLeft"/>
  61. /// </summary>
  62. public int AttemptsLeft { get; private set;}
  63. /// <summary>
  64. /// <inheritdoc cref="ITaskItem.InitialAttepts"/>
  65. /// </summary>
  66. public int InitialAttempts { get; private set; }
  67. /// <summary>
  68. /// <inheritdoc cref="ITaskItem.AttemptTimeout"/>
  69. /// </summary>
  70. public TimeSpan? AttemptTimeout { get; private set;}
  71. /// <summary>
  72. /// <inheritdoc cref="ITaskItem.StatStarted"/>
  73. /// </summary>
  74. public DateTime StatStarted { get; private set;}
  75. /// <summary>
  76. /// <inheritdoc cref="ITaskItem.StatFinished"/>
  77. /// </summary>
  78. public DateTime StatFinished { get; private set;}
  79. /// <summary>
  80. /// <inheritdoc cref="ITaskItem.StatLastAttemptDuration"/>
  81. /// </summary>
  82. public TimeSpan StatLastAttemptDuration { get; private set;}
  83. /// <summary>
  84. /// <inheritdoc cref="ITaskItem.StatusLog"/>
  85. /// </summary>
  86. public ITaskStatusLogItem[] StatusLog => _statusLog.ToArray();
  87. public TypedArgument[] Arguments => _arguments.ToArray();
  88. #endregion
  89. #region *** Private Fields ***
  90. private readonly List<ITaskStatusLogItem> _statusLog = new();
  91. private readonly List<TypedArgument> _arguments = new();
  92. private ILog _log;
  93. private ProcessQueue _parent;
  94. private Thread _thread;
  95. private CancellationTokenSource _cancellationTokenSource;
  96. #endregion
  97. #region *** Constructors ***
  98. private TaskItem(ProcessQueue parent, long id, ProcessStatusEnum initialStatus, string reference, string source, TypedArgument[] initialArguments, byte priority, int attepts, TimeSpan? attemptTimeout, DateTime validFrom, DateTime? validTo, ITaskStatusLogItem[]? initialStatusLog):
  99. this(parent, reference, source, initialArguments, priority, attepts, attemptTimeout, validFrom, validTo, initialStatusLog)
  100. {
  101. Id = id;
  102. SetStatus(initialStatus, "(System) Initialization");
  103. }
  104. public TaskItem(ProcessQueue parent, string reference, string providerClass, TypedArgument[] initialArguments, byte priority, int attepts, TimeSpan? attemptTimeout, DateTime validFrom, DateTime? validTo, ITaskStatusLogItem[]? initialStatusLog)
  105. {
  106. AffinityToken = null;
  107. _parent = parent ?? throw new ArgumentNullException(nameof(parent));
  108. _log = new ConsoleLogger().GetLogger(GetType());
  109. Priority = priority;
  110. Reference = reference;
  111. ProviderClass = providerClass;
  112. _arguments.Setup(initialArguments);
  113. AttemptTimeout = attemptTimeout;
  114. AttemptsLeft = attepts;
  115. InitialAttempts = attepts;
  116. ProcessingValidFrom = validFrom;
  117. ProcessingValidTo = validTo;
  118. Created = DateTime.Now;
  119. if (initialStatusLog != null)
  120. _statusLog.Setup(initialStatusLog);
  121. SetStatus(ProcessStatusEnum.Pending, "(System) Initialization");
  122. Id = long.MinValue;
  123. }
  124. #endregion
  125. #region *** Public Operations ***
  126. /// <summary>
  127. /// Starts task and sets <see cref="Status"/> to <see cref="ProcessStatusEnum.Attached"/> then <see cref="ProcessStatusEnum.Started"/> or
  128. /// <see cref="ProcessStatusEnum.ProcessedSucc"/> or <see cref="ProcessStatusEnum.ProcessedFail"/> (target status is depends on input arguments)..
  129. /// <list type="bullet">
  130. /// <item><description>Allowed state: <see cref="ProcessStatusEnum.Pending"/> or <see cref="ProcessStatusEnum.Paused"/></description></item>
  131. /// <item><description>Set state to: <see cref="ProcessStatusEnum.Fa"/> or <see cref="ProcessStatusEnum.ProcessedSucc"/> or <see cref="ProcessStatusEnum.ProcessedFail"/></description></item>
  132. /// </list>
  133. /// <exception cref="InvalidTaskStateException">When task state is invalid for operation.</exception>
  134. /// </summary>
  135. /// <param name="reason">Reason message to stop.</param>
  136. /// <param name="willPause">If true, then <see cref="Status"/> will be set to <see cref="ProcessStatusEnum.Paused"/> otherwise to <see cref="ProcessStatusEnum.ProcessedSucc"/> or <see cref="ProcessStatusEnum.ProcessedFail"/>.</param>
  137. /// <param name="willFail">If true, then <see cref="Status"/> will be set to <see cref="ProcessStatusEnum.ProcessedFail"/> otherwise to <see cref="ProcessStatusEnum.ProcessedSucc"/>.</param>
  138. public void Start(IAffinityToken token)
  139. {
  140. if (_cancellationTokenSource != null) throw new InvalidOperationException("Inconsistent task state. Thread must not be initialized before task start.");
  141. Reload();
  142. CheckTaskState(ProcessStatusEnum.Pending, ProcessStatusEnum.Paused);
  143. if (AttemptsLeft == 0)
  144. {
  145. SetStatus(ProcessStatusEnum.ExpiredAttemts, "(System) Attempts expired");
  146. _parent.Provider.Set(this);
  147. return;
  148. }
  149. AffinityToken = token;
  150. SetStatus(ProcessStatusEnum.Attached, $"Attached to worker {AffinityToken}");
  151. _parent.Provider.Set(this);
  152. _cancellationTokenSource = new CancellationTokenSource();
  153. if (AttemptTimeout.HasValue)
  154. _cancellationTokenSource.CancelAfter(AttemptTimeout.Value);
  155. ThreadPool.QueueUserWorkItem(Execute, _cancellationTokenSource);
  156. }
  157. public void Postpone(TimeSpan postponeInterval, string reason)
  158. {
  159. throw new NotImplementedException();
  160. }
  161. /// <summary>
  162. /// <inheritdoc cref="ITaskItem.Stop"/>
  163. /// </summary>
  164. public void Stop(string reason, bool willPause = true, bool willFail = false)
  165. {
  166. if (_cancellationTokenSource == null) throw new InvalidOperationException("Inconsistent task state. Thread must be initialized before task stop.");
  167. Reload();
  168. CheckTaskState(ProcessStatusEnum.Started);
  169. if (AttemptsLeft == 0)
  170. {
  171. SetStatus(ProcessStatusEnum.ExpiredAttemts, "(System) Attempts expired");
  172. _parent.Provider.Set(this);
  173. return;
  174. }
  175. _cancellationTokenSource.Cancel();
  176. _cancellationTokenSource = null;
  177. SetStatus(willPause ? ProcessStatusEnum.Paused : willFail ? ProcessStatusEnum.ProcessedFail : ProcessStatusEnum.ProcessedSucc, reason);
  178. StatFinished = DateTime.Now;
  179. _parent.Provider.Set(this);
  180. }
  181. public void Reset(string reason, TimeSpan postponeValid)
  182. {
  183. throw new NotImplementedException();
  184. }
  185. public void Delete(string reason, bool immediate = false)
  186. {
  187. throw new NotImplementedException();
  188. }
  189. public bool Euqals(ITaskItemData data)
  190. {
  191. return ((ITaskItemData)this).Equals(data);
  192. }
  193. public override string ToString()
  194. {
  195. var sb = new StringBuilder();
  196. sb.Append("Id=");
  197. sb.Append(Id == long.MinValue ? "NEW" : Id.ToString()).Append(";");
  198. sb.Append("Ref=").Append(Reference).Append(";");
  199. sb.Append("Sta=").Append(Status).Append(";");
  200. sb.Append("Cre=").Append(Created.ToString("G")).Append(";");
  201. return sb.ToString();
  202. }
  203. #endregion
  204. #region *** Internal Operations ***
  205. internal void SetStatus(ProcessStatusEnum status, string reason)
  206. {
  207. if (status == Status)
  208. {
  209. var log = _statusLog.Where(x => x.Status == status).MaxBy(x => x.Created);
  210. if (log == null) throw new InvalidOperationException($"Inconsistent task state. Status log must not be empty for status '{status}'.");
  211. if (log.Reason != reason)
  212. {
  213. ((TaskStatusLogItem)log).Reason = reason;
  214. Changed = DateTime.Now;
  215. return;
  216. }
  217. }
  218. _statusLog.Add(new TaskStatusLogItem(status, reason));
  219. switch (status)
  220. {
  221. case ProcessStatusEnum.Initial:
  222. Created = DateTime.Now;
  223. break;
  224. case ProcessStatusEnum.Started:
  225. StatStarted = DateTime.Now;
  226. break;
  227. case ProcessStatusEnum.ProcessedFail:
  228. case ProcessStatusEnum.ProcessedSucc:
  229. case ProcessStatusEnum.ExpiredAttemts:
  230. case ProcessStatusEnum.ExpiredValidity:
  231. case ProcessStatusEnum.Failed:
  232. StatFinished = DateTime.Now;
  233. break;
  234. }
  235. Status = status;
  236. Changed = DateTime.Now;
  237. }
  238. internal void Reload()
  239. {
  240. if (Id == long.MinValue) throw new InvalidOperationException("Task is not saved yet.");
  241. var data = _parent.GetTask(this);
  242. CopyFrom(data);
  243. }
  244. internal void CopyFrom(ITaskItemData data)
  245. {
  246. Id = data.Id;
  247. Status = data.Status;
  248. Changed = data.Changed;
  249. Reference = data.Reference;
  250. ProviderClass = data.ProviderClass;
  251. Priority = data.Priority;
  252. ProcessingValidFrom = data.ProcessingValidFrom;
  253. ProcessingValidTo = data.ProcessingValidTo;
  254. Created = data.Created;
  255. Changed = data.Changed;
  256. StatStarted = data.StatStarted;
  257. StatFinished = data.StatFinished;
  258. StatLastAttemptDuration = data.StatLastAttemptDuration;
  259. AttemptsLeft = data.AttemptsLeft;
  260. InitialAttempts = data.InitialAttempts;
  261. AttemptTimeout = data.AttemptTimeout;
  262. _statusLog.Clear();
  263. _statusLog.AddRange(data.StatusLog);
  264. _arguments.Clear();
  265. _arguments.AddRange(data.Arguments);
  266. }
  267. bool IEquatable<ITaskItemData>.Equals(ITaskItemData? other)
  268. {
  269. if (other == null)
  270. return false;
  271. var result = Id == other.Id &&
  272. Status == other.Status &&
  273. Changed == other.Changed &&
  274. Reference == other.Reference &&
  275. ProviderClass == other.ProviderClass &&
  276. Priority == other.Priority &&
  277. ProcessingValidFrom == other.ProcessingValidFrom &&
  278. ProcessingValidTo == other.ProcessingValidTo &&
  279. Created == other.Created &&
  280. Changed == other.Changed &&
  281. StatStarted == other.StatStarted &&
  282. StatFinished == other.StatFinished &&
  283. StatLastAttemptDuration == other.StatLastAttemptDuration &&
  284. InitialAttempts == other.InitialAttempts &&
  285. AttemptsLeft == other.AttemptsLeft &&
  286. AttemptTimeout == other.AttemptTimeout &&
  287. _statusLog.SequenceEqual(other.StatusLog) &&
  288. _arguments.SequenceEqual(other.Arguments);
  289. return result;
  290. }
  291. #endregion
  292. #region *** Private Operations ***
  293. private void CheckValidity()
  294. {
  295. var now = DateTime.Now;
  296. if (!(ProcessingValidFrom <= now && now < ProcessingValidTo.GetValueOrDefault(DateTime.MaxValue)))
  297. throw new TaskValidityExpiredException(this);
  298. }
  299. private void CheckTaskState(params ProcessStatusEnum[] allowedStates)
  300. {
  301. if (allowedStates == null || allowedStates.Length == 0) throw new ArgumentNullException(nameof(allowedStates));
  302. CheckValidity();
  303. if (!allowedStates.Contains(Status))
  304. throw new InvalidTaskStateException(this, allowedStates);
  305. }
  306. private void Execute(object? state)
  307. {
  308. try
  309. {
  310. if (state==null) throw new ArgumentNullException(nameof(state));
  311. var token = (CancellationTokenSource)state;
  312. SetStatus(ProcessStatusEnum.Started, $"(System) Task started with attempt #{(InitialAttempts - AttemptsLeft) + 1}");
  313. StatStarted = DateTime.Now;
  314. _parent.Provider.Set(this);
  315. }
  316. catch (Exception e)
  317. {
  318. }
  319. }
  320. #endregion
  321. }
  322. }