.net工作流elsa-触发器 - ToB企服应用市场:ToB评测及商务社交产业平台

protected override IEnumerable<object> GetTriggerPayloads(TriggerIndexingContext context) => GetBookmarkPayloads(context.ExpressionExecutionContext);
private IEnumerable<object> GetBookmarkPayloads(ExpressionExecutionContext context)
{
// Generate bookmark data for path and selected methods.
var normalizedRoute = context.Get(Path)!.NormalizeRoute();
var methods = SupportedMethods.GetOrDefault(context) ?? new List<string> { HttpMethods.Get };
var authorize = Authorize.GetOrDefault(context);
var policy = Policy.GetOrDefault(context);
var requestTimeout = RequestTimeout.GetOrDefault(context);
var requestSizeLimit = RequestSizeLimit.GetOrDefault(context);
//根据http请求方法，返回多个数据，会在触发器索引列表中创建多条记录
return methods
.Select(x => new HttpEndpointBookmarkPayload(normalizedRoute, x.ToLowerInvariant(), authorize, policy, requestTimeout, requestSizeLimit))
.Cast<object>()
.ToArray();
}

复制代码

public class UpdateRouteTable(IRouteTableUpdater routeTableUpdater, IOptions<HttpActivityOptions> options) :
INotificationHandler<WorkflowTriggersIndexed>,
INotificationHandler<WorkflowBookmarksIndexed>
{
/// <inheritdoc />
public async Task HandleAsync(WorkflowTriggersIndexed notification, CancellationToken cancellationToken)
{
routeTableUpdater.RemoveRoutes(notification.IndexedWorkflowTriggers.RemovedTriggers);
await routeTableUpdater.AddRoutesAsync(notification.IndexedWorkflowTriggers.AddedTriggers, cancellationToken);
await routeTableUpdater.AddRoutesAsync(notification.IndexedWorkflowTriggers.UnchangedTriggers, cancellationToken);
}

复制代码

public async Task InvokeAsync(HttpContext httpContext, IServiceProvider serviceProvider)
{
//当前请求路径
var path = GetPath(httpContext);
//根据elsa路由表匹配路由数据
var matchingPath = GetMatchingRoute(serviceProvider, path).Route;
//配置elsa时，HttpActivityOptions中指定的基础地址
var basePath = options.Value.BasePath?.ToString().NormalizeRoute();
//若请求地址连elsa配置的基础地址都不匹配，则直接执行下个中间件，说明没见听到触发器定义的要求
// If the request path does not match the configured base path to handle workflows, then skip.
if (!string.IsNullOrWhiteSpace(basePath))
{
if (!path.StartsWith(basePath, StringComparison.OrdinalIgnoreCase))
{
await next(httpContext);
return;
}
// Strip the base path.
matchingPath = matchingPath[basePath.Length..];
}
matchingPath = matchingPath.NormalizeRoute();
var input = new Dictionary<string, object>
{
[HttpEndpoint.HttpContextInputKey] = true,
[HttpEndpoint.RequestPathInputKey] = path.NormalizeRoute()
};
var cancellationToken = httpContext.RequestAborted;
var request = httpContext.Request;
var method = request.Method.ToLowerInvariant();
var httpWorkflowLookupService = serviceProvider.GetRequiredService<IHttpWorkflowLookupService>();
var workflowInstanceId = await GetWorkflowInstanceIdAsync(serviceProvider, httpContext, cancellationToken);
var correlationId = await GetCorrelationIdAsync(serviceProvider, httpContext, cancellationToken);
//根据请求路径 http方法和 HttpEndpoint计算hash值
var bookmarkHash = ComputeBookmarkHash(serviceProvider, matchingPath, method);
//根据上面的hash值，从存储中获取匹配的工作流及其触发器列表
var lookupResult = await httpWorkflowLookupService.FindWorkflowAsync(bookmarkHash, cancellationToken);
if (lookupResult != null)
{
//若找到了流程，且里面仅包含一个与当前请求匹配的触发器，则说了匹配上了，执行流程，否则报错
var triggers = lookupResult.Triggers;
if (triggers.Count > 1)
{
//报错
await HandleMultipleWorkflowsFoundAsync(httpContext, () => triggers.Select(x => new
{
x.WorkflowDefinitionId
}), cancellationToken);
return;
}
var trigger = triggers.FirstOrDefault();
if (trigger != null)
{
var workflowGraph = lookupResult.WorkflowGraph!;
//执行流程中触发器所在节点
await StartWorkflowAsync(httpContext, trigger, workflowGraph, input, workflowInstanceId, correlationId);
return;
}
}
//若触发器节点已经执行过，也就是之前从其它节点流转过去的，那次触发器节点被执行时会创建书签的，则直接根据书签恢复执行
var bookmarks = await FindBookmarksAsync(serviceProvider, bookmarkHash, workflowInstanceId, correlationId, cancellationToken).ToList();
//若找到多个匹配的，报错
if (bookmarks.Count > 1)
{
await HandleMultipleWorkflowsFoundAsync(httpContext, () => bookmarks.Select(x => new
{
x.WorkflowInstanceId
}), cancellationToken);
return;
}
var bookmark = bookmarks.SingleOrDefault();
if (bookmark != null)
{
//恢复书签执行
await ResumeWorkflowAsync(httpContext, bookmark, input, correlationId);
return;
}
// 如果基础地址都匹配上了，却没找到对应的流程，则抛出404错误
if (basePath != null)
{
await httpContext.Response.SendNotFoundAsync(cancellation: cancellationToken);
return;
}
// If no base path was configured, the request should be handled by subsequent middlewares.
await next(httpContext);
}

复制代码

protected override async ValueTask ExecuteAsync(ActivityExecutionContext context)
{
var path = Path.Get(context);
if (path.Contains("//"))
throw new RoutePatternException(path, "Path cannot contain double slashes (//)");
//如果本次执行不是由当前HttpEndpoint自己触发的，比如在当前节点前另一个触发器执行了此流程，但是节点流转到这里来了。
if (!context.IsTriggerOfWorkflow())
{
//则直接创建书签卡住流程，等到匹配的http请求被HttpWorkflowsMiddleware流转到这里时，OnResumeAsync将被执行
context.CreateBookmarks(GetBookmarkPayloads(context.ExpressionExecutionContext), includeActivityInstanceId: false, callback: OnResumeAsync);
return;
}
//否则，说明当前流程的执行，就是这里配置的触发器触发的。
var httpContextAccessor = context.GetRequiredService<IHttpContextAccessor>();
var httpContext = httpContextAccessor.HttpContext;
//触发器也是activity，它可能直接被执行，而不是被http请求执行，则需要等待传统的书签请求来恢复OnResumeAsync
if (httpContext == null)
{
// We're executing in a non-HTTP context (e.g. in a virtual actor).
// Create a bookmark to allow the invoker to export the state and resume execution from there.
context.CreateBookmark(OnResumeAsync, BookmarkMetadata.HttpCrossBoundary);
return;
}
//否则说明当前流程的触发器就是自己，并且被http请求触发了
await HandleRequestAsync(context, httpContext);
}

复制代码

private async ValueTask OnResumeAsync(ActivityExecutionContext context)
{
var httpContextAccessor = context.GetRequiredService<IHttpContextAccessor>();
var httpContext = httpContextAccessor.HttpContext;
//在恢复执行时，可能并不是http请求恢复的，可能是直接调用书签恢复的
if (httpContext == null)
{
// We're executing in a non-HTTP context (e.g. in a virtual actor).
// Create a bookmark to allow the invoker to export the state and resume execution from there.
context.CreateBookmark(OnResumeAsync, BookmarkMetadata.HttpCrossBoundary);
return;
}
//处理http请求
await HandleRequestAsync(context, httpContext);
}

复制代码

public async Task ScheduleAsync(IEnumerable<StoredTrigger> triggers, CancellationToken cancellationToken = default)
{
var triggerList = triggers.ToList();
var timerTriggers = triggerList.Filter<Activities.Timer>();
var startAtTriggers = triggerList.Filter<StartAt>();
var cronTriggers = triggerList.Filter<Cron>();
var now = systemClock.UtcNow;
// Schedule each Timer trigger.
foreach (var trigger in timerTriggers)
{
var (startAt, interval) = trigger.GetPayload<TimerTriggerPayload>();
var input = new { StartAt = startAt, Interval = interval }.ToDictionary();
//安排流程作业时，要求创建新的流程实例
var request = new ScheduleNewWorkflowInstanceRequest
{
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionVersionId(trigger.WorkflowDefinitionVersionId),
TriggerActivityId = trigger.ActivityId,
Input = input
};
await workflowScheduler.ScheduleRecurringAsync(trigger.Id, request, startAt, interval, cancellationToken);
}
// Schedule each StartAt trigger.
foreach (var trigger in startAtTriggers)
{
var executeAt = trigger.GetPayload<StartAtPayload>().ExecuteAt;
// If the trigger is in the past, log info and skip scheduling.
if (executeAt < now)
{
logger.LogInformation("StartAt trigger is in the past. TriggerId: {TriggerId}. ExecuteAt: {ExecuteAt}. Skipping scheduling", trigger.Id, executeAt);
continue;
}
var input = new { ExecuteAt = executeAt }.ToDictionary();
//安排流程作业时，要求创建新的流程实例
var request = new ScheduleNewWorkflowInstanceRequest
{
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionVersionId(trigger.WorkflowDefinitionVersionId),
TriggerActivityId = trigger.ActivityId,
Input = input
};
await workflowScheduler.ScheduleAtAsync(trigger.Id, request, executeAt, cancellationToken);
}
// Schedule each Cron trigger.
foreach (var trigger in cronTriggers)
{
var payload = trigger.GetPayload<CronTriggerPayload>();
var cronExpression = payload.CronExpression;
if (string.IsNullOrWhiteSpace(cronExpression))
{
logger.LogWarning("Cron expression is empty. TriggerId: {TriggerId}. Skipping scheduling of this trigger", trigger.Id);
continue;
}
var input = new { CronExpression = cronExpression }.ToDictionary();
//安排流程作业时，要求创建新的流程实例
var request = new ScheduleNewWorkflowInstanceRequest
{
WorkflowDefinitionHandle = WorkflowDefinitionHandle.ByDefinitionVersionId(trigger.WorkflowDefinitionVersionId),
TriggerActivityId = trigger.ActivityId,
Input = input
};
try
{
await workflowScheduler.ScheduleCronAsync(trigger.Id, request, cronExpression, cancellationToken);
}
catch (FormatException ex)
{
logger.LogWarning(ex, "Cron expression format error. CronExpression: {CronExpression}", cronExpression);
}
}
}

复制代码

public class ScheduleWorkflows : INotificationHandler<WorkflowTriggersIndexed>, INotificationHandler<WorkflowBookmarksIndexed>
{
//...其它代码
public async Task HandleAsync(WorkflowTriggersIndexed notification, CancellationToken cancellationToken)
{
//注销之前的后台作业
await _triggerScheduler.UnscheduleAsync(notification.IndexedWorkflowTriggers.RemovedTriggers, cancellationToken);
//使用后台作业，在指定时间点恢复触发器所在节点
await _triggerScheduler.ScheduleAsync(notification.IndexedWorkflowTriggers.AddedTriggers, cancellationToken);
}

复制代码