SemianiakaVY/elsa-core
1
0
Fork 0
Code Issues Pull requests Actions 1 Packages Projects Releases Wiki Activity

Release distributed lock

Browse source
This commit is contained in:
Sipke Schoorstra 2021-01-08 22:00:44 +01:00
parent 0f8d6b6442
commit 409bf36e7e
1 changed files with 25 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
src/activities/Elsa.Activities.Timers.Quartz/Jobs/RunQuartzWorkflowJob.cs
View file

@ -1,4 +1,6 @@
using System.Threading.Tasks;
using System;
using System.Diagnostics;
using System.Threading.Tasks;
using Elsa.DistributedLock;
using Elsa.Models;
using Elsa.Persistence;
@ -16,6 +18,7 @@ namespace Elsa.Activities.Timers.Quartz.Jobs
private readonly IWorkflowQueue _workflowQueue;
private readonly IDistributedLockProvider _distributedLockProvider;
private readonly ILogger _logger;
private Stopwatch _stopwatch = new();
public RunQuartzWorkflowJob(
IWorkflowRegistry workflowRegistry,
@ -41,22 +44,36 @@ namespace Elsa.Activities.Timers.Quartz.Jobs
var activityId = dataMap.GetString("ActivityId")!;
var lockKey = (workflowInstanceId, workflowDefinitionId, activityId).GetHashCode().ToString();
_logger.LogDebug("Acquiring lock on {WorkflowInstanceId} / {WorkflowDefinitionId} / {ActivityId}", workflowInstanceId, workflowDefinitionId, activityId);
if (!await _distributedLockProvider.AcquireLockAsync(lockKey, cancellationToken))
{
_logger.LogDebug("Failed to acquire lock on {WorkflowInstanceId} / {WorkflowDefinitionId} / {ActivityId}", workflowInstanceId, workflowDefinitionId, activityId);
return;
}
var workflowBlueprint = (await _workflowRegistry.GetWorkflowAsync(workflowDefinitionId, tenantId, VersionOptions.Published, cancellationToken))!;
_stopwatch.Restart();
try
{
var workflowBlueprint = (await _workflowRegistry.GetWorkflowAsync(workflowDefinitionId, tenantId, VersionOptions.Published, cancellationToken))!;
if (workflowInstanceId == null)
{
if (workflowBlueprint.IsSingleton || await GetWorkflowIsAlreadyExecutingAsync(tenantId, workflowDefinitionId) == false)
await _workflowQueue.EnqueueWorkflowDefinition(workflowDefinitionId, tenantId, activityId, null, null, null, cancellationToken);
if (workflowInstanceId == null)
{
if (workflowBlueprint.IsSingleton || await GetWorkflowIsAlreadyExecutingAsync(tenantId, workflowDefinitionId) == false)
await _workflowQueue.EnqueueWorkflowDefinition(workflowDefinitionId, tenantId, activityId, null, null, null, cancellationToken);
}
else
{
await _workflowQueue.EnqueueWorkflowInstance(workflowInstanceId, activityId, null, cancellationToken);
}
}
else
finally
{
await _workflowQueue.EnqueueWorkflowInstance(workflowInstanceId, activityId, null, cancellationToken);
_stopwatch.Stop();
await _distributedLockProvider.ReleaseLockAsync(lockKey, cancellationToken);
_logger.LogDebug("Held lock on {WorkflowInstanceId} / {WorkflowDefinitionId} / {ActivityId} for {LockTime}.", workflowInstanceId, workflowDefinitionId, activityId, _stopwatch.Elapsed);
}
}