Merge pull request danielgerlag#475 from cocytus/utc-now

danielgerlag · web-flow · commit 61ccea7a10c2 · 2019-12-18T19:18:17.000-08:00
Add UtcNow to IDateTimeProvider, and use that throughout.
diff --git a/src/WorkflowCore/Interface/IDateTimeProvider.cs b/src/WorkflowCore/Interface/IDateTimeProvider.cs
@@ -5,5 +5,6 @@ namespace WorkflowCore.Interface
     public interface IDateTimeProvider
     {
         DateTime Now { get; }
+        DateTime UtcNow { get; }
     }
 }
diff --git a/src/WorkflowCore/Services/BackgroundTasks/EventConsumer.cs b/src/WorkflowCore/Services/BackgroundTasks/EventConsumer.cs
@@ -37,7 +37,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance
             {
                 cancellationToken.ThrowIfCancellationRequested();
                 var evt = await _persistenceStore.GetEvent(itemId);
-                if (evt.EventTime <= _datetimeProvider.Now.ToUniversalTime())
+                if (evt.EventTime <= _datetimeProvider.UtcNow)
                 {
                     var subs = await _persistenceStore.GetSubcriptions(evt.EventName, evt.EventKey, evt.EventTime);
                     var toQueue = new List<string>();
diff --git a/src/WorkflowCore/Services/BackgroundTasks/WorkflowConsumer.cs b/src/WorkflowCore/Services/BackgroundTasks/WorkflowConsumer.cs
@@ -71,7 +71,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance
 
                         await persistenceStore.PersistErrors(result.Errors);
 
-                        var readAheadTicks = _datetimeProvider.Now.Add(Options.PollInterval).ToUniversalTime().Ticks;
+                        var readAheadTicks = _datetimeProvider.UtcNow.Add(Options.PollInterval).Ticks;
 
                         if ((workflow.Status == WorkflowStatus.Runnable) && workflow.NextExecution.HasValue && workflow.NextExecution.Value < readAheadTicks)
                         {
@@ -109,7 +109,7 @@ private async void FutureQueue(WorkflowInstance workflow, CancellationToken canc
                     return;
                 }
 
-                var target = (workflow.NextExecution.Value - _datetimeProvider.Now.ToUniversalTime().Ticks);
+                var target = (workflow.NextExecution.Value - _datetimeProvider.UtcNow.Ticks);
                 if (target > 0)
                 {
                     await Task.Delay(TimeSpan.FromTicks(target), cancellationToken);
diff --git a/src/WorkflowCore/Services/DateTimeProvider.cs b/src/WorkflowCore/Services/DateTimeProvider.cs
@@ -6,5 +6,6 @@ namespace WorkflowCore.Services
     public class DateTimeProvider : IDateTimeProvider
     {
         public DateTime Now => DateTime.Now;
+        public DateTime UtcNow => DateTime.UtcNow;
     }
 }
diff --git a/src/WorkflowCore/Services/ErrorHandlers/CompensateHandler.cs b/src/WorkflowCore/Services/ErrorHandlers/CompensateHandler.cs
@@ -60,7 +60,7 @@ public void Handle(WorkflowInstance workflow, WorkflowDefinition def, ExecutionP
                 }
 
                 scopePointer.Active = false;
-                scopePointer.EndTime = _datetimeProvider.Now.ToUniversalTime();
+                scopePointer.EndTime = _datetimeProvider.UtcNow;
                 scopePointer.Status = PointerStatus.Failed;
 
                 if (scopeStep.CompensationStepId.HasValue)
diff --git a/src/WorkflowCore/Services/ErrorHandlers/RetryHandler.cs b/src/WorkflowCore/Services/ErrorHandlers/RetryHandler.cs
@@ -22,7 +22,7 @@ public RetryHandler(IDateTimeProvider datetimeProvider, WorkflowOptions options)
         public void Handle(WorkflowInstance workflow, WorkflowDefinition def, ExecutionPointer pointer, WorkflowStep step, Exception exception, Queue<ExecutionPointer> bubbleUpQueue)
         {
             pointer.RetryCount++;
-            pointer.SleepUntil = _datetimeProvider.Now.ToUniversalTime().Add(step.RetryInterval ?? def.DefaultErrorRetryInterval ?? _options.ErrorRetryInterval);
+            pointer.SleepUntil = _datetimeProvider.UtcNow.Add(step.RetryInterval ?? def.DefaultErrorRetryInterval ?? _options.ErrorRetryInterval);
             step.PrimeForRetry(pointer);
         }
     }
diff --git a/src/WorkflowCore/Services/ErrorHandlers/SuspendHandler.cs b/src/WorkflowCore/Services/ErrorHandlers/SuspendHandler.cs
@@ -24,7 +24,7 @@ public void Handle(WorkflowInstance workflow, WorkflowDefinition def, ExecutionP
             workflow.Status = WorkflowStatus.Suspended;
             _eventPublisher.PublishNotification(new WorkflowSuspended()
             {
-                EventTimeUtc = _datetimeProvider.Now,
+                EventTimeUtc = _datetimeProvider.UtcNow,
                 Reference = workflow.Reference,
                 WorkflowInstanceId = workflow.Id,
                 WorkflowDefinitionId = workflow.WorkflowDefinitionId,
diff --git a/src/WorkflowCore/Services/ErrorHandlers/TerminateHandler.cs b/src/WorkflowCore/Services/ErrorHandlers/TerminateHandler.cs
@@ -24,7 +24,7 @@ public void Handle(WorkflowInstance workflow, WorkflowDefinition def, ExecutionP
             workflow.Status = WorkflowStatus.Terminated;
             _eventPublisher.PublishNotification(new WorkflowTerminated()
             {
-                EventTimeUtc = _datetimeProvider.Now,
+                EventTimeUtc = _datetimeProvider.UtcNow,
                 Reference = workflow.Reference,
                 WorkflowInstanceId = workflow.Id,
                 WorkflowDefinitionId = workflow.WorkflowDefinitionId,
diff --git a/src/WorkflowCore/Services/ExecutionResultProcessor.cs b/src/WorkflowCore/Services/ExecutionResultProcessor.cs
@@ -33,7 +33,7 @@ public void ProcessExecutionResult(WorkflowInstance workflow, WorkflowDefinition
             pointer.Outcome = result.OutcomeValue;
             if (result.SleepFor.HasValue)
             {
-                pointer.SleepUntil = _datetimeProvider.Now.ToUniversalTime().Add(result.SleepFor.Value);
+                pointer.SleepUntil = _datetimeProvider.UtcNow.Add(result.SleepFor.Value);
                 pointer.Status = PointerStatus.Sleeping;
             }
 
@@ -57,7 +57,7 @@ public void ProcessExecutionResult(WorkflowInstance workflow, WorkflowDefinition
             if (result.Proceed)
             {
                 pointer.Active = false;
-                pointer.EndTime = _datetimeProvider.Now.ToUniversalTime();
+                pointer.EndTime = _datetimeProvider.UtcNow;
                 pointer.Status = PointerStatus.Complete;
 
                 foreach (var outcomeTarget in step.Outcomes.Where(x => object.Equals(x.GetValue(workflow.Data), result.OutcomeValue) || x.GetValue(workflow.Data) == null))
@@ -67,7 +67,7 @@ public void ProcessExecutionResult(WorkflowInstance workflow, WorkflowDefinition
 
                 _eventPublisher.PublishNotification(new StepCompleted()
                 {
-                    EventTimeUtc = _datetimeProvider.Now,
+                    EventTimeUtc = _datetimeProvider.UtcNow,
                     Reference = workflow.Reference,
                     ExecutionPointerId = pointer.Id,
                     StepId = step.Id,
@@ -92,7 +92,7 @@ public void HandleStepException(WorkflowInstance workflow, WorkflowDefinition de
         {
             _eventPublisher.PublishNotification(new WorkflowError()
             {
-                EventTimeUtc = _datetimeProvider.Now,
+                EventTimeUtc = _datetimeProvider.UtcNow,
                 Reference = workflow.Reference,
                 WorkflowInstanceId = workflow.Id,
                 WorkflowDefinitionId = workflow.WorkflowDefinitionId,
diff --git a/src/WorkflowCore/Services/WorkflowExecutor.cs b/src/WorkflowCore/Services/WorkflowExecutor.cs
@@ -41,7 +41,7 @@ public async Task<WorkflowExecutorResult> Execute(WorkflowInstance workflow)
         {
             var wfResult = new WorkflowExecutorResult();
 
-            var exePointers = new List<ExecutionPointer>(workflow.ExecutionPointers.Where(x => x.Active && (!x.SleepUntil.HasValue || x.SleepUntil < _datetimeProvider.Now.ToUniversalTime())));
+            var exePointers = new List<ExecutionPointer>(workflow.ExecutionPointers.Where(x => x.Active && (!x.SleepUntil.HasValue || x.SleepUntil < _datetimeProvider.UtcNow)));
             var def = _registry.GetDefinition(workflow.WorkflowDefinitionId, workflow.Version);
             if (def == null)
             {
@@ -60,12 +60,12 @@ public async Task<WorkflowExecutorResult> Execute(WorkflowInstance workflow)
                 if (step == null)
                 {
                     _logger.LogError("Unable to find step {0} in workflow definition", pointer.StepId);
-                    pointer.SleepUntil = _datetimeProvider.Now.ToUniversalTime().Add(_options.ErrorRetryInterval);
+                    pointer.SleepUntil = _datetimeProvider.UtcNow.Add(_options.ErrorRetryInterval);
                     wfResult.Errors.Add(new ExecutionError()
                     {
                         WorkflowId = workflow.Id,
                         ExecutionPointerId = pointer.Id,
-                        ErrorTime = _datetimeProvider.Now.ToUniversalTime(),
+                        ErrorTime = _datetimeProvider.UtcNow,
                         Message = $"Unable to find step {pointer.StepId} in workflow definition"
                     });
                     continue;
@@ -85,7 +85,7 @@ public async Task<WorkflowExecutorResult> Execute(WorkflowInstance workflow)
                     {
                         WorkflowId = workflow.Id,
                         ExecutionPointerId = pointer.Id,
-                        ErrorTime = _datetimeProvider.Now.ToUniversalTime(),
+                        ErrorTime = _datetimeProvider.UtcNow,
                         Message = ex.Message
                     });
                         
@@ -108,7 +108,7 @@ private bool InitializeStep(WorkflowInstance workflow, WorkflowStep step, Workfl
                     return false;
                 case ExecutionPipelineDirective.EndWorkflow:
                     workflow.Status = WorkflowStatus.Complete;
-                    workflow.CompleteTime = _datetimeProvider.Now.ToUniversalTime();
+                    workflow.CompleteTime = _datetimeProvider.UtcNow;
                     return false;
             }
 
@@ -117,7 +117,7 @@ private bool InitializeStep(WorkflowInstance workflow, WorkflowStep step, Workfl
                 pointer.Status = PointerStatus.Running;
                 _publisher.PublishNotification(new StepStarted()
                 {
-                    EventTimeUtc = _datetimeProvider.Now,
+                    EventTimeUtc = _datetimeProvider.UtcNow,
                     Reference = workflow.Reference,
                     ExecutionPointerId = pointer.Id,
                     StepId = step.Id,
@@ -129,7 +129,7 @@ private bool InitializeStep(WorkflowInstance workflow, WorkflowStep step, Workfl
 
             if (!pointer.StartTime.HasValue)
             {
-                pointer.StartTime = _datetimeProvider.Now.ToUniversalTime();
+                pointer.StartTime = _datetimeProvider.UtcNow;
             }
 
             return true;
@@ -146,12 +146,12 @@ private async Task ExecuteStep(WorkflowInstance workflow, WorkflowStep step, Exe
                 if (body == null)
                 {
                     _logger.LogError("Unable to construct step body {0}", step.BodyType.ToString());
-                    pointer.SleepUntil = _datetimeProvider.Now.ToUniversalTime().Add(_options.ErrorRetryInterval);
+                    pointer.SleepUntil = _datetimeProvider.UtcNow.Add(_options.ErrorRetryInterval);
                     wfResult.Errors.Add(new ExecutionError()
                     {
                         WorkflowId = workflow.Id,
                         ExecutionPointerId = pointer.Id,
-                        ErrorTime = _datetimeProvider.Now.ToUniversalTime(),
+                        ErrorTime = _datetimeProvider.UtcNow,
                         Message = $"Unable to construct step body {step.BodyType.ToString()}"
                     });
                     return;
@@ -175,7 +175,7 @@ private async Task ExecuteStep(WorkflowInstance workflow, WorkflowStep step, Exe
                         return;
                     case ExecutionPipelineDirective.EndWorkflow:
                         workflow.Status = WorkflowStatus.Complete;
-                        workflow.CompleteTime = _datetimeProvider.Now.ToUniversalTime();
+                        workflow.CompleteTime = _datetimeProvider.UtcNow;
                         return;
                 }
 
@@ -245,10 +245,10 @@ private void DetermineNextExecutionTime(WorkflowInstance workflow)
                 return;
             
             workflow.Status = WorkflowStatus.Complete;
-            workflow.CompleteTime = _datetimeProvider.Now.ToUniversalTime();
+            workflow.CompleteTime = _datetimeProvider.UtcNow;
             _publisher.PublishNotification(new WorkflowCompleted()
             {
-                EventTimeUtc = _datetimeProvider.Now,
+                EventTimeUtc = _datetimeProvider.UtcNow,
                 Reference = workflow.Reference,
                 WorkflowInstanceId = workflow.Id,
                 WorkflowDefinitionId = workflow.WorkflowDefinitionId,
diff --git a/test/WorkflowCore.UnitTests/Services/ExecutionResultProcessorFixture.cs b/test/WorkflowCore.UnitTests/Services/ExecutionResultProcessorFixture.cs
@@ -34,6 +34,7 @@ public ExecutionResultProcessorFixture()
             Options = new WorkflowOptions(A.Fake<IServiceCollection>());
 
             A.CallTo(() => DateTimeProvider.Now).Returns(DateTime.Now);
+            A.CallTo(() => DateTimeProvider.UtcNow).Returns(DateTime.UtcNow);
 
             //config logging
             var loggerFactory = new LoggerFactory();
diff --git a/test/WorkflowCore.UnitTests/Services/WorkflowExecutorFixture.cs b/test/WorkflowCore.UnitTests/Services/WorkflowExecutorFixture.cs
@@ -48,6 +48,7 @@ public WorkflowExecutorFixture()
             A.CallTo(() => scope.ServiceProvider).Returns(ServiceProvider);
 
             A.CallTo(() => DateTimeProvider.Now).Returns(DateTime.Now);
+            A.CallTo(() => DateTimeProvider.UtcNow).Returns(DateTime.UtcNow);
 
             //config logging
             var loggerFactory = new LoggerFactory();

Original file line number	Diff line number	Diff line change
`@@ -5,5 +5,6 @@ namespace WorkflowCore.Interface`
`5`	`5`	`public interface IDateTimeProvider`
`6`	`6`	`{`
`7`	`7`	`DateTime Now { get; }`
	`8`	`+ DateTime UtcNow { get; }`
`8`	`9`	`}`
`9`	`10`	`}`
Original file line number	Diff line number	Diff line change
`@@ -37,7 +37,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance`
`37`	`37`	`{`
`38`	`38`	`cancellationToken.ThrowIfCancellationRequested();`
`39`	`39`	`var evt = await _persistenceStore.GetEvent(itemId);`
`40`		`- if (evt.EventTime <= _datetimeProvider.Now.ToUniversalTime())`
	`40`	`+ if (evt.EventTime <= _datetimeProvider.UtcNow)`
`41`	`41`	`{`
`42`	`42`	`var subs = await _persistenceStore.GetSubcriptions(evt.EventName, evt.EventKey, evt.EventTime);`
`43`	`43`	`var toQueue = new List<string>();`
Original file line number	Diff line number	Diff line change
`@@ -71,7 +71,7 @@ protected override async Task ProcessItem(string itemId, CancellationToken cance`
`71`	`71`
`72`	`72`	`await persistenceStore.PersistErrors(result.Errors);`
`73`	`73`
`74`		`- var readAheadTicks = _datetimeProvider.Now.Add(Options.PollInterval).ToUniversalTime().Ticks;`
	`74`	`+ var readAheadTicks = _datetimeProvider.UtcNow.Add(Options.PollInterval).Ticks;`
`75`	`75`
`76`	`76`	`if ((workflow.Status == WorkflowStatus.Runnable) && workflow.NextExecution.HasValue && workflow.NextExecution.Value < readAheadTicks)`
`77`	`77`	`{`
`@@ -109,7 +109,7 @@ private async void FutureQueue(WorkflowInstance workflow, CancellationToken canc`
`109`	`109`	`return;`
`110`	`110`	`}`
`111`	`111`
`112`		`- var target = (workflow.NextExecution.Value - _datetimeProvider.Now.ToUniversalTime().Ticks);`
	`112`	`+ var target = (workflow.NextExecution.Value - _datetimeProvider.UtcNow.Ticks);`
`113`	`113`	`if (target > 0)`
`114`	`114`	`{`
`115`	`115`	`await Task.Delay(TimeSpan.FromTicks(target), cancellationToken);`
Original file line number	Diff line number	Diff line change
`@@ -6,5 +6,6 @@ namespace WorkflowCore.Services`
`6`	`6`	`public class DateTimeProvider : IDateTimeProvider`
`7`	`7`	`{`
`8`	`8`	`public DateTime Now => DateTime.Now;`
	`9`	`+ public DateTime UtcNow => DateTime.UtcNow;`
`9`	`10`	`}`
`10`	`11`	`}`
Original file line number	Diff line number	Diff line change
`@@ -60,7 +60,7 @@ public void Handle(WorkflowInstance workflow, WorkflowDefinition def, ExecutionP`
`60`	`60`	`}`
`61`	`61`
`62`	`62`	`scopePointer.Active = false;`
`63`		`- scopePointer.EndTime = _datetimeProvider.Now.ToUniversalTime();`
	`63`	`+ scopePointer.EndTime = _datetimeProvider.UtcNow;`
`64`	`64`	`scopePointer.Status = PointerStatus.Failed;`
`65`	`65`
`66`	`66`	`if (scopeStep.CompensationStepId.HasValue)`
Original file line number	Diff line number	Diff line change
`@@ -22,7 +22,7 @@ public RetryHandler(IDateTimeProvider datetimeProvider, WorkflowOptions options)`
`22`	`22`	`public void Handle(WorkflowInstance workflow, WorkflowDefinition def, ExecutionPointer pointer, WorkflowStep step, Exception exception, Queue<ExecutionPointer> bubbleUpQueue)`
`23`	`23`	`{`
`24`	`24`	`pointer.RetryCount++;`
`25`		`- pointer.SleepUntil = _datetimeProvider.Now.ToUniversalTime().Add(step.RetryInterval ?? def.DefaultErrorRetryInterval ?? _options.ErrorRetryInterval);`
	`25`	`+ pointer.SleepUntil = _datetimeProvider.UtcNow.Add(step.RetryInterval ?? def.DefaultErrorRetryInterval ?? _options.ErrorRetryInterval);`
`26`	`26`	`step.PrimeForRetry(pointer);`
`27`	`27`	`}`
`28`	`28`	`}`
Original file line number	Diff line number	Diff line change
`@@ -24,7 +24,7 @@ public void Handle(WorkflowInstance workflow, WorkflowDefinition def, ExecutionP`
`24`	`24`	`workflow.Status = WorkflowStatus.Suspended;`
`25`	`25`	`_eventPublisher.PublishNotification(new WorkflowSuspended()`
`26`	`26`	`{`
`27`		`- EventTimeUtc = _datetimeProvider.Now,`
	`27`	`+ EventTimeUtc = _datetimeProvider.UtcNow,`
`28`	`28`	`Reference = workflow.Reference,`
`29`	`29`	`WorkflowInstanceId = workflow.Id,`
`30`	`30`	`WorkflowDefinitionId = workflow.WorkflowDefinitionId,`
Original file line number	Diff line number	Diff line change
`@@ -33,7 +33,7 @@ public void ProcessExecutionResult(WorkflowInstance workflow, WorkflowDefinition`
`33`	`33`	`pointer.Outcome = result.OutcomeValue;`
`34`	`34`	`if (result.SleepFor.HasValue)`
`35`	`35`	`{`
`36`		`- pointer.SleepUntil = _datetimeProvider.Now.ToUniversalTime().Add(result.SleepFor.Value);`
	`36`	`+ pointer.SleepUntil = _datetimeProvider.UtcNow.Add(result.SleepFor.Value);`
`37`	`37`	`pointer.Status = PointerStatus.Sleeping;`
`38`	`38`	`}`
`39`	`39`
`@@ -57,7 +57,7 @@ public void ProcessExecutionResult(WorkflowInstance workflow, WorkflowDefinition`
`57`	`57`	`if (result.Proceed)`
`58`	`58`	`{`
`59`	`59`	`pointer.Active = false;`
`60`		`- pointer.EndTime = _datetimeProvider.Now.ToUniversalTime();`
	`60`	`+ pointer.EndTime = _datetimeProvider.UtcNow;`
`61`	`61`	`pointer.Status = PointerStatus.Complete;`
`62`	`62`
`63`	`63`	`foreach (var outcomeTarget in step.Outcomes.Where(x => object.Equals(x.GetValue(workflow.Data), result.OutcomeValue) \|\| x.GetValue(workflow.Data) == null))`
`@@ -67,7 +67,7 @@ public void ProcessExecutionResult(WorkflowInstance workflow, WorkflowDefinition`
`67`	`67`
`68`	`68`	`_eventPublisher.PublishNotification(new StepCompleted()`
`69`	`69`	`{`
`70`		`- EventTimeUtc = _datetimeProvider.Now,`
	`70`	`+ EventTimeUtc = _datetimeProvider.UtcNow,`
`71`	`71`	`Reference = workflow.Reference,`
`72`	`72`	`ExecutionPointerId = pointer.Id,`
`73`	`73`	`StepId = step.Id,`
`@@ -92,7 +92,7 @@ public void HandleStepException(WorkflowInstance workflow, WorkflowDefinition de`
`92`	`92`	`{`
`93`	`93`	`_eventPublisher.PublishNotification(new WorkflowError()`
`94`	`94`	`{`
`95`		`- EventTimeUtc = _datetimeProvider.Now,`
	`95`	`+ EventTimeUtc = _datetimeProvider.UtcNow,`
`96`	`96`	`Reference = workflow.Reference,`
`97`	`97`	`WorkflowInstanceId = workflow.Id,`
`98`	`98`	`WorkflowDefinitionId = workflow.WorkflowDefinitionId,`