84 lines
2.9 KiB
C#
84 lines
2.9 KiB
C#
|
using System.Globalization;
|
||
|
|
using System.Text.Json;
|
||
|
|
using System.Text.Json.Serialization;
|
||
|
|
using w4c_workflows.Models;
|
||
|
|
|
||
|
|
namespace w4c_workflows.Services.Execution;
|
||
|
|
|
||
|
|
/// <summary>
|
||
|
|
/// Serializes a <c>task.run</c> job from the compiled <see cref="WorkflowTask"/>
|
||
|
|
/// into the flat Redis Stream fields the worker understands (see
|
||
|
|
/// <see cref="TaskInvocation.FromFields"/>). This is the control-plane side of
|
||
|
|
/// the wire contract; the worker side deserializes the same field names.
|
||
|
|
/// </summary>
|
||
|
|
public static class TaskRunMessage
|
||
|
|
{
|
||
|
|
private static readonly JsonSerializerOptions Json = new(JsonSerializerDefaults.Web)
|
||
|
|
{
|
||
|
|
DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
|
||
|
|
};
|
||
|
|
|
||
|
|
/// <summary>
|
||
|
|
/// Builds the stream fields for executing <paramref name="task"/> as part of
|
||
|
|
/// <paramref name="run"/>. <paramref name="workingDir"/> is the directory
|
||
|
|
/// containing the task's code files; the entry file is taken from the task's
|
||
|
|
/// <c>EntryJson</c> and stays relative to it.
|
||
|
|
/// </summary>
|
||
|
|
public static IReadOnlyDictionary<string, string> ToFields(
|
||
|
|
WorkflowRun run,
|
||
|
|
WorkflowTask task,
|
||
|
|
string? input,
|
||
|
|
string? workingDir,
|
||
|
|
int attempt)
|
||
|
|
{
|
||
|
|
var entry = ParseEntry(task.EntryJson);
|
||
|
|
|
||
|
|
var fields = new Dictionary<string, string>
|
||
|
|
{
|
||
|
|
["type"] = TaskInvocation.TypeValue,
|
||
|
|
["run_id"] = run.Id.ToString(),
|
||
|
|
["task_id"] = task.Id.ToString(),
|
||
|
|
["task_key"] = task.Key,
|
||
|
|
["language"] = task.Language,
|
||
|
|
["entry_file"] = entry.File ?? string.Empty,
|
||
|
|
["tenant_id"] = run.TenantId,
|
||
|
|
["attempt"] = attempt.ToString(CultureInfo.InvariantCulture),
|
||
|
|
};
|
||
|
|
|
||
|
|
if (!string.IsNullOrWhiteSpace(entry.Function))
|
||
|
|
fields["entry_function"] = entry.Function;
|
||
|
|
|
||
|
|
var env = DeserializeEnv(task.EnvJson);
|
||
|
|
if (env.Count > 0)
|
||
|
|
fields["env"] = JsonSerializer.Serialize(env, Json);
|
||
|
|
|
||
|
|
if (input != null)
|
||
|
|
fields["input"] = input;
|
||
|
|
if (!string.IsNullOrWhiteSpace(workingDir))
|
||
|
|
fields["working_dir"] = workingDir;
|
||
|
|
if (!string.IsNullOrWhiteSpace(task.Server))
|
||
|
|
fields["server"] = task.Server!;
|
||
|
|
|
||
|
|
return fields;
|
||
|
|
}
|
||
|
|
|
||
|
|
private static EntryDefinition ParseEntry(string? entryJson)
|
||
|
|
{
|
||
|
|
if (!string.IsNullOrWhiteSpace(entryJson))
|
||
|
|
{
|
||
|
|
var entry = JsonSerializer.Deserialize<EntryDefinition>(entryJson, Json);
|
||
|
|
if (entry != null)
|
||
|
|
return entry;
|
||
|
|
}
|
||
|
|
return new EntryDefinition { File = string.Empty };
|
||
|
|
}
|
||
|
|
|
||
|
|
private static IReadOnlyDictionary<string, string> DeserializeEnv(string? envJson)
|
||
|
|
{
|
||
|
|
if (string.IsNullOrWhiteSpace(envJson))
|
||
|
|
return new Dictionary<string, string>();
|
||
|
|
return JsonSerializer.Deserialize<Dictionary<string, string>>(envJson, Json)
|
||
|
|
?? new Dictionary<string, string>();
|
||
|
|
}
|
||
|
|
}
|