behzadalizadeh/Caspian.Elsa.Core
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

InitAsync

Browse Source
This commit is contained in:
Behzad Alizadeh
2026-01-01 20:00:02 +03:30
parent 2c3b98cd24
commit 1759d2f21f
2 changed files with 176 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/apps/Caspian.WorkflowFramework/Services/WorkflowService.cs
+170 -13
View File
@@ -1,28 +1,20 @@
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Diagnostics;
using System.Dynamic;
using System.Linq;
using System.Text;
using System.Text.Json;
using System.Threading.Tasks;
using Azure;
using Caspian.WorkflowFramework.Contracts;
using Caspian.WorkflowFramework.Models;
using Caspian.WorkflowFramework.Utility;
using Elsa.Common.Models;
using Elsa.JavaScript.Handlers;
using Elsa.Workflows;
using Elsa.Workflows.Management;
using Elsa.Workflows.Management.Filters;
using Elsa.Workflows.Models;
using Elsa.Workflows.Runtime;
using Elsa.Workflows.Runtime.Messages;
using FastEndpoints;
using Microsoft.CodeAnalysis.Operations;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Win32;
using Newtonsoft.Json;
using Newtonsoft.Json.Converters;
using Proto.Extensions;
using San.MongoLogger;
namespace Caspian.WorkflowFramework.Services;
@@ -35,6 +27,7 @@ public class WorkflowService : IWorkflowService, ILoggable
private readonly IWorkflowInstanceVariableManager _workflowInstanceVariableManager;
private readonly IWorkflowInstanceManager _workflowInstanceManager;
private readonly IEventPublisher _eventPublisher;
private readonly IWorkflowDefinitionCacheManager _workflowDefinitionCacheManager;
private readonly IMdbLogger<WorkflowService> _logger;
#endregion
@@ -139,6 +132,101 @@ public class WorkflowService : IWorkflowService, ILoggable
}
#endregion
#region InitAsync
/// <inheritdoc />
public async Task<WorkflowResponse<WorkflowModel>> InitAsync(CreateAndExecuteWorkflowModel input)
{
var sw = Stopwatch.StartNew();
try
{
_logger.Log("Start of InitAsync", input);
WorkflowResponse<WorkflowModel> response;
var workflowClient = await _workflowRuntime.CreateClientAsync();
var definition = await _workflowDefinitionService
.FindWorkflowDefinitionAsync(new WorkflowDefinitionFilter()
{
Name = input.Name
});
if (definition == null)
{
var msg = $"Workflow '{input.Name}' defintion not found";
response = new WorkflowResponse<WorkflowModel>()
{
Status = 1,
Message = msg,
Data = null
};
sw.Stop();
_logger.Log("End of InitAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds);
return response;
}
var runningWorkflowInstance = await _workflowInstanceManager
.FindAsync(new WorkflowInstanceFilter()
{
ExternalUserId = input.UserId,
DefinitionId = "jhzbsddvjkf",//definition.DefinitionId,
WorkflowStatus = WorkflowStatus.Running
});
var instanceInput = new Dictionary<string, object> { ["UserId"] = input.UserId };
input.Version = VersionOptions.Published;
var newInstance = await workflowClient.CreateInstanceAsync(new CreateWorkflowInstanceRequest
{
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionId(definition.DefinitionId, input.Version),
Input = instanceInput,
ExternalUserId = input.UserId
});
var result = await workflowClient.RunInstanceAsync(RunWorkflowInstanceRequest.Empty);
var variables = await _workflowInstanceVariableManager.GetVariablesAsync(result.WorkflowInstanceId);
// تبدیل به دیکشنری
var dictionary = variables.ToDictionary(a => a.Variable.Name, a => a.Value);
var workflowModel = dictionary.ConvertToWorkflowModel();
workflowModel.InstanceId = workflowClient.WorkflowInstanceId;
workflowModel.UserId = input.UserId;
response = new WorkflowResponse<WorkflowModel>()
{
Status = 0,
Message = "Success",
Data = workflowModel
};
sw.Stop();
_logger.Log("End of InitAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds);
return response;
}
catch (Exception ex)
{
var response = new WorkflowResponse<WorkflowModel>()
{
Status = 100,
Message = ex.Message,
Data = null
};
sw.Stop();
_logger.Log("Exception occurs in InitAsync", WFInstanceId: response.Data?.InstanceId, response, sw.ElapsedMilliseconds, ex);
return response;
}
}
#endregion
#region PublishEventAsync
/// <inheritdoc />
public async Task<WorkflowResponse<WorkflowModel>> PublishEventAsync(PublishEventModel input)
@@ -239,7 +327,14 @@ public class WorkflowService : IWorkflowService, ILoggable
WorkflowResponse<WorkflowModel> response;
var instance = await _workflowInstanceManager.FindByIdAsync(input.WorkflowInstanceId);
var instance = await _workflowInstanceManager.FindAsync(new WorkflowInstanceFilter()
{
ExternalUserId = 62181149,
DefinitionId = "8a44ce8bc5c71839",
WorkflowStatus = Elsa.Workflows.WorkflowStatus.Running
});
//var instance = await _workflowInstanceManager.FindByIdAsync(input.WorkflowInstanceId);
if (instance == null)
{
@@ -258,7 +353,7 @@ public class WorkflowService : IWorkflowService, ILoggable
return response;
}
var variables = await _workflowInstanceVariableManager.GetVariablesAsync(input.WorkflowInstanceId);
var variables = await _workflowInstanceVariableManager.GetVariablesAsync(instance.Id);// input.WorkflowInstanceId);
// تبدیل به دیکشنری
var dictionary = variables.ToDictionary(a => a.Variable.Name, a => a.Value);
@@ -297,4 +392,66 @@ public class WorkflowService : IWorkflowService, ILoggable
}
}
#endregion
//#region EvictDefFromCacheAsync
///// <inheritdoc />
//public async Task<WorkflowResponse<WorkflowModel>> EvictDefFromCacheAsync(string definitionId)
//{
// var sw = Stopwatch.StartNew();
// try
// {
// _logger.Log("Start of EvictDefFromCacheAsync", definitionId);
// var key = _workflowDefinitionCacheManager.CreateWorkflowDefinitionChangeTokenKey(definitionId);
// await _workflowDefinitionCacheManager.EvictWorkflowDefinitionAsync(definitionId);
// WorkflowResponse<WorkflowModel> response;
// if (instance == null)
// {
// var msg = $"Workflow instance with id '{input.WorkflowInstanceId}' not found";
// response = new WorkflowResponse<WorkflowModel>()
// {
// Status = 1,
// Message = msg,
// Data = null
// };
// sw.Stop();
// _logger.Log("End of InquiryStateAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds);
// return response;
// }
// response = new WorkflowResponse<WorkflowModel>()
// {
// Status = 0,
// Message = "Success",
// Data = workflowModel
// };
// sw.Stop();
// _logger.Log("End of InquiryStateAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds);
// return response;
// }
// catch (Exception ex)
// {
// var response = new WorkflowResponse<WorkflowModel>()
// {
// Status = 100,
// Message = ex.Message,
// Data = null
// };
// sw.Stop();
// _logger.Log("Exception occurs in InquiryStateAsync", WFInstanceId: input.WorkflowInstanceId, response, sw.ElapsedMilliseconds, ex);
// return response;
// }
//}
//#endregion
}
src/modules/Elsa.EntityFrameworkCore/Modules/Management/WorkflowInstanceStore.cs
+6 -4
View File
@@ -50,7 +50,9 @@ public class EFCoreWorkflowInstanceStore : IWorkflowInstanceStore
/// <inheritdoc />
public async ValueTask<WorkflowInstance?> FindAsync(WorkflowInstanceFilter filter, CancellationToken cancellationToken = default)
{
return await _store.QueryAsync(query => Filter(query, filter), OnLoadAsync, cancellationToken).FirstOrDefault();
return await _store
.QueryAsync(query => Filter(query, filter).OrderByDescending(x=>x.CreatedAt), OnLoadAsync, cancellationToken)
.FirstOrDefault();
}
/// <inheritdoc />
@@ -177,10 +179,10 @@ public class EFCoreWorkflowInstanceStore : IWorkflowInstanceStore
{
var proposedValues = entry.CurrentValues;
var databaseValues = await entry.GetDatabaseValuesAsync(cancellationToken);
if(databaseValues == null)
if (databaseValues == null)
continue;
var updatedAtProperty = entry.Metadata.GetProperty(nameof(WorkflowInstance.UpdatedAt));
var proposedValue = (DateTimeOffset)proposedValues[updatedAtProperty]!;
var databaseValue = (DateTimeOffset)databaseValues[updatedAtProperty]!;