| | | 1 | | using Elsa.Common; |
| | | 2 | | using Elsa.Common.Codecs; |
| | | 3 | | using Elsa.Workflows.LogPersistence; |
| | | 4 | | using Elsa.Workflows.Management.Options; |
| | | 5 | | using Elsa.Workflows.Runtime.Entities; |
| | | 6 | | using Elsa.Workflows.Runtime.Extensions; |
| | | 7 | | using Elsa.Workflows.State; |
| | | 8 | | using Microsoft.Extensions.Options; |
| | | 9 | | |
| | | 10 | | namespace Elsa.Workflows.Runtime; |
| | | 11 | | |
| | | 12 | | /// <inheritdoc /> |
| | 514 | 13 | | public class DefaultActivityExecutionMapper( |
| | 514 | 14 | | ISafeSerializer safeSerializer, |
| | 514 | 15 | | IPayloadSerializer payloadSerializer, |
| | 514 | 16 | | ICompressionCodecResolver compressionCodecResolver, |
| | 514 | 17 | | IOptions<ManagementOptions> options) : IActivityExecutionMapper |
| | | 18 | | { |
| | | 19 | | /// <inheritdoc /> |
| | | 20 | | public async Task<ActivityExecutionRecord> MapAsync(ActivityExecutionContext source) |
| | | 21 | | { |
| | 4244 | 22 | | var outputs = source.GetOutputs(); |
| | 4244 | 23 | | var inputs = source.GetInputs(); |
| | 4244 | 24 | | var persistenceMap = source.GetLogPersistenceModeMap(); |
| | 4244 | 25 | | var persistableInputs = GetPersistableInputOutput(inputs, persistenceMap.Inputs); |
| | 4244 | 26 | | var persistableOutputs = GetPersistableInputOutput(outputs, persistenceMap.Outputs); |
| | 4244 | 27 | | var persistableProperties = GetPersistableDictionary(source.Properties!, persistenceMap.InternalState); |
| | 4244 | 28 | | var persistableJournalData = GetPersistableDictionary(source.JournalData!, persistenceMap.InternalState); |
| | 4244 | 29 | | var cancellationToken = source.CancellationToken; |
| | | 30 | | |
| | 4244 | 31 | | var record = new ActivityExecutionRecord |
| | 4244 | 32 | | { |
| | 4244 | 33 | | Id = source.Id, |
| | 4244 | 34 | | ActivityId = source.Activity.Id, |
| | 4244 | 35 | | ActivityNodeId = source.NodeId, |
| | 4244 | 36 | | WorkflowInstanceId = source.WorkflowExecutionContext.Id, |
| | 4244 | 37 | | ActivityType = source.Activity.Type, |
| | 4244 | 38 | | ActivityName = source.Activity.Name, |
| | 4244 | 39 | | ActivityState = persistableInputs, |
| | 4244 | 40 | | Outputs = persistableOutputs, |
| | 4244 | 41 | | Properties = persistableProperties!, |
| | 4244 | 42 | | Metadata = new Dictionary<string, object>(source.Metadata), |
| | 4244 | 43 | | Payload = persistableJournalData!, |
| | 4244 | 44 | | Exception = ExceptionState.FromException(source.Exception), |
| | 4244 | 45 | | ActivityTypeVersion = source.Activity.Version, |
| | 4244 | 46 | | StartedAt = source.StartedAt, |
| | 4244 | 47 | | HasBookmarks = source.Bookmarks.Any(), |
| | 4244 | 48 | | Status = source.Status, |
| | 4244 | 49 | | AggregateFaultCount = source.AggregateFaultCount, |
| | 4244 | 50 | | CompletedAt = source.CompletedAt |
| | 4244 | 51 | | }; |
| | | 52 | | |
| | 4244 | 53 | | record = record.SanitizeLogMessage(); |
| | 4244 | 54 | | var compressionAlgorithm = options.Value.CompressionAlgorithm ?? nameof(None); |
| | 4244 | 55 | | var serializedActivityState = record.ActivityState?.Count > 0 ? safeSerializer.Serialize(record.ActivityState) : |
| | 4244 | 56 | | var compressedSerializedActivityState = serializedActivityState != null ? await compressionCodecResolver.Resolve |
| | 4244 | 57 | | var serializedProperties = record.Properties != null ? payloadSerializer.Serialize(record.Properties) : null; |
| | 4244 | 58 | | var serializedMetadata = record.Metadata != null ? payloadSerializer.Serialize(record.Metadata) : null; |
| | 4244 | 59 | | record.SerializedSnapshot = new() |
| | 4244 | 60 | | { |
| | 4244 | 61 | | Id = record.Id, |
| | 4244 | 62 | | TenantId = record.TenantId, |
| | 4244 | 63 | | WorkflowInstanceId = record.WorkflowInstanceId, |
| | 4244 | 64 | | ActivityId = record.ActivityId, |
| | 4244 | 65 | | ActivityNodeId = record.ActivityNodeId, |
| | 4244 | 66 | | ActivityType = record.ActivityType, |
| | 4244 | 67 | | ActivityTypeVersion = record.ActivityTypeVersion, |
| | 4244 | 68 | | ActivityName = record.ActivityName, |
| | 4244 | 69 | | StartedAt = record.StartedAt, |
| | 4244 | 70 | | HasBookmarks = record.HasBookmarks, |
| | 4244 | 71 | | Status = record.Status, |
| | 4244 | 72 | | AggregateFaultCount = record.AggregateFaultCount, |
| | 4244 | 73 | | CompletedAt = record.CompletedAt, |
| | 4244 | 74 | | SerializedActivityState = compressedSerializedActivityState, |
| | 4244 | 75 | | SerializedActivityStateCompressionAlgorithm = compressionAlgorithm, |
| | 4244 | 76 | | SerializedOutputs = record.Outputs?.Any() == true ? safeSerializer.Serialize(record.Outputs) : null, |
| | 4244 | 77 | | SerializedProperties = serializedProperties, |
| | 4244 | 78 | | SerializedMetadata = serializedMetadata, |
| | 4244 | 79 | | SerializedException = record.Exception != null ? payloadSerializer.Serialize(record.Exception) : null, |
| | 4244 | 80 | | SerializedPayload = record.Payload?.Any() == true ? payloadSerializer.Serialize(record.Payload) : null |
| | 4244 | 81 | | }; |
| | | 82 | | |
| | 4244 | 83 | | return record; |
| | 4244 | 84 | | } |
| | | 85 | | |
| | | 86 | | private IDictionary<string, object?> GetPersistableInputOutput(IDictionary<string, object> state, IDictionary<string |
| | | 87 | | { |
| | 8488 | 88 | | var result = new Dictionary<string, object?>(); |
| | 34836 | 89 | | foreach (var stateEntry in state) |
| | | 90 | | { |
| | 8930 | 91 | | var mode = map.TryGetValue(stateEntry.Key, out var value) ? value : LogPersistenceMode.Include; |
| | 8930 | 92 | | if (mode == LogPersistenceMode.Include) |
| | 8922 | 93 | | result.Add(stateEntry.Key, stateEntry.Value); |
| | | 94 | | } |
| | | 95 | | |
| | 8488 | 96 | | return result; |
| | | 97 | | } |
| | | 98 | | |
| | | 99 | | private IDictionary<string, object?>? GetPersistableDictionary(IDictionary<string, object?> dictionary, LogPersisten |
| | | 100 | | { |
| | 8488 | 101 | | return mode == LogPersistenceMode.Include ? new Dictionary<string, object?>(dictionary) : null; |
| | | 102 | | } |
| | | 103 | | } |