diff --git a/src/api/Synapse.Api.Application/Synapse.Api.Application.csproj b/src/api/Synapse.Api.Application/Synapse.Api.Application.csproj index 5a56b155..c817e12b 100644 --- a/src/api/Synapse.Api.Application/Synapse.Api.Application.csproj +++ b/src/api/Synapse.Api.Application/Synapse.Api.Application.csproj @@ -7,7 +7,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/api/Synapse.Api.Client.Core/Synapse.Api.Client.Core.csproj b/src/api/Synapse.Api.Client.Core/Synapse.Api.Client.Core.csproj index 038a3046..245da389 100644 --- a/src/api/Synapse.Api.Client.Core/Synapse.Api.Client.Core.csproj +++ b/src/api/Synapse.Api.Client.Core/Synapse.Api.Client.Core.csproj @@ -7,7 +7,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/api/Synapse.Api.Client.Http/Synapse.Api.Client.Http.csproj b/src/api/Synapse.Api.Client.Http/Synapse.Api.Client.Http.csproj index b0083eb2..6351a39b 100644 --- a/src/api/Synapse.Api.Client.Http/Synapse.Api.Client.Http.csproj +++ b/src/api/Synapse.Api.Client.Http/Synapse.Api.Client.Http.csproj @@ -7,7 +7,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/api/Synapse.Api.Http/Synapse.Api.Http.csproj b/src/api/Synapse.Api.Http/Synapse.Api.Http.csproj index 06d31262..ecfca4e5 100644 --- a/src/api/Synapse.Api.Http/Synapse.Api.Http.csproj +++ b/src/api/Synapse.Api.Http/Synapse.Api.Http.csproj @@ -8,7 +8,7 @@ Library True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/api/Synapse.Api.Server/Synapse.Api.Server.csproj b/src/api/Synapse.Api.Server/Synapse.Api.Server.csproj index 093aad1a..4dba39d6 100644 --- a/src/api/Synapse.Api.Server/Synapse.Api.Server.csproj +++ b/src/api/Synapse.Api.Server/Synapse.Api.Server.csproj @@ -7,7 +7,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/cli/Synapse.Cli/Synapse.Cli.csproj b/src/cli/Synapse.Cli/Synapse.Cli.csproj index 70079e45..e1f97ab7 100644 --- a/src/cli/Synapse.Cli/Synapse.Cli.csproj +++ b/src/cli/Synapse.Cli/Synapse.Cli.csproj @@ -8,7 +8,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/core/Synapse.Core.Infrastructure.Containers.Docker/Synapse.Core.Infrastructure.Containers.Docker.csproj b/src/core/Synapse.Core.Infrastructure.Containers.Docker/Synapse.Core.Infrastructure.Containers.Docker.csproj index b39a6548..a275615a 100644 --- a/src/core/Synapse.Core.Infrastructure.Containers.Docker/Synapse.Core.Infrastructure.Containers.Docker.csproj +++ b/src/core/Synapse.Core.Infrastructure.Containers.Docker/Synapse.Core.Infrastructure.Containers.Docker.csproj @@ -7,7 +7,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/core/Synapse.Core.Infrastructure.Containers.Kubernetes/Synapse.Core.Infrastructure.Containers.Kubernetes.csproj b/src/core/Synapse.Core.Infrastructure.Containers.Kubernetes/Synapse.Core.Infrastructure.Containers.Kubernetes.csproj index 4d5fa673..130724dd 100644 --- a/src/core/Synapse.Core.Infrastructure.Containers.Kubernetes/Synapse.Core.Infrastructure.Containers.Kubernetes.csproj +++ b/src/core/Synapse.Core.Infrastructure.Containers.Kubernetes/Synapse.Core.Infrastructure.Containers.Kubernetes.csproj @@ -7,7 +7,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/core/Synapse.Core.Infrastructure/Synapse.Core.Infrastructure.csproj b/src/core/Synapse.Core.Infrastructure/Synapse.Core.Infrastructure.csproj index a2ee6abe..29b76edb 100644 --- a/src/core/Synapse.Core.Infrastructure/Synapse.Core.Infrastructure.csproj +++ b/src/core/Synapse.Core.Infrastructure/Synapse.Core.Infrastructure.csproj @@ -7,7 +7,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/core/Synapse.Core/Synapse.Core.csproj b/src/core/Synapse.Core/Synapse.Core.csproj index 76c0e3b6..c256c332 100644 --- a/src/core/Synapse.Core/Synapse.Core.csproj +++ b/src/core/Synapse.Core/Synapse.Core.csproj @@ -7,7 +7,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/correlator/Synapse.Correlator/Synapse.Correlator.csproj b/src/correlator/Synapse.Correlator/Synapse.Correlator.csproj index af3a358d..6fc97f54 100644 --- a/src/correlator/Synapse.Correlator/Synapse.Correlator.csproj +++ b/src/correlator/Synapse.Correlator/Synapse.Correlator.csproj @@ -8,7 +8,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/operator/Synapse.Operator/Synapse.Operator.csproj b/src/operator/Synapse.Operator/Synapse.Operator.csproj index e11fe155..93d7712e 100644 --- a/src/operator/Synapse.Operator/Synapse.Operator.csproj +++ b/src/operator/Synapse.Operator/Synapse.Operator.csproj @@ -8,7 +8,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/runner/Synapse.Runner/Services/Executors/ListenTaskExecutor.cs b/src/runner/Synapse.Runner/Services/Executors/ListenTaskExecutor.cs index 5c6f617f..90231273 100644 --- a/src/runner/Synapse.Runner/Services/Executors/ListenTaskExecutor.cs +++ b/src/runner/Synapse.Runner/Services/Executors/ListenTaskExecutor.cs @@ -62,7 +62,13 @@ protected override async Task DoExecuteAsync(CancellationToken cancellationToken if (this.Task.Definition.Foreach == null) { var context = await this.Task.CorrelateAsync(cancellationToken).ConfigureAwait(false); - await this.SetResultAsync(context, this.Task.Definition.Then, cancellationToken).ConfigureAwait(false); + var events = this.Task.Definition.Listen.Read switch + { + EventReadMode.Data or EventReadMode.Raw => context.Events.Select(e => e.Value.Data), + EventReadMode.Envelope => context.Events.Select(e => e.Value.Data), + _ => throw new NotSupportedException($"The specified event read mode '{this.Task.Definition.Listen.Read}' is not supported") + }; + await this.SetResultAsync(events, this.Task.Definition.Then, cancellationToken).ConfigureAwait(false); } else { @@ -107,7 +113,12 @@ protected virtual async Task OnStreamingEventAsync(IStreamedCloudEvent e) ] }; var arguments = this.GetExpressionEvaluationArguments(); - var eventData = e.Event as object; + var eventData = this.Task.Definition.Listen.Read switch + { + EventReadMode.Data or EventReadMode.Raw => e.Event.Data, + EventReadMode.Envelope => e.Event, + _ => throw new NotSupportedException($"The specified event read mode '{this.Task.Definition.Listen.Read}' is not supported") + }; if (this.Task.Definition.Foreach.Output?.As is string fromExpression) eventData = await this.Task.Workflow.Expressions.EvaluateAsync(fromExpression, eventData ?? new(), arguments, this.CancellationTokenSource!.Token).ConfigureAwait(false); else if (this.Task.Definition.Foreach.Output?.As != null) eventData = await this.Task.Workflow.Expressions.EvaluateAsync(this.Task.Definition.Foreach.Output.As!, eventData ?? new(), arguments, this.CancellationTokenSource!.Token).ConfigureAwait(false); if (this.Task.Definition.Foreach.Export?.As is string toExpression) diff --git a/src/runner/Synapse.Runner/Services/TaskExecutor.cs b/src/runner/Synapse.Runner/Services/TaskExecutor.cs index 78c7687c..e655edc6 100644 --- a/src/runner/Synapse.Runner/Services/TaskExecutor.cs +++ b/src/runner/Synapse.Runner/Services/TaskExecutor.cs @@ -379,20 +379,6 @@ public virtual async Task SkipAsync(object? result, string? then = FlowDirective this.Stopwatch.Stop(); if (string.IsNullOrWhiteSpace(then)) then = FlowDirective.Continue; var output = result; - var arguments = this.GetExpressionEvaluationArguments() ?? new Dictionary(); - arguments[RuntimeExpressions.Arguments.Output] = output!; - if (this.Task.Definition.Output?.As is string fromExpression) output = await this.Task.Workflow.Expressions.EvaluateAsync(fromExpression, output ?? new(), arguments, cancellationToken).ConfigureAwait(false); - else if (this.Task.Definition.Output?.As != null) output = await this.Task.Workflow.Expressions.EvaluateAsync(this.Task.Definition.Output.As, output ?? new(), arguments, cancellationToken).ConfigureAwait(false); - if (this.Task.Definition.Export?.As is string toExpression) - { - var context = (await this.Task.Workflow.Expressions.EvaluateAsync>(toExpression, this.Task.ContextData, arguments, cancellationToken).ConfigureAwait(false))!; - await this.Task.SetContextDataAsync(context, cancellationToken).ConfigureAwait(false); - } - else if (this.Task.Definition.Export?.As != null) - { - var context = (await this.Task.Workflow.Expressions.EvaluateAsync>(this.Task.Definition.Export.As, this.Task.ContextData, this.GetExpressionEvaluationArguments(), cancellationToken).ConfigureAwait(false))!; - await this.Task.SetContextDataAsync(context, cancellationToken).ConfigureAwait(false); - } await this.Task.SkipAsync(output, then, cancellationToken).ConfigureAwait(false); this.Subject.OnNext(new TaskLifeCycleEvent(TaskLifeCycleEventType.Skipped)); this.Subject.OnCompleted(); diff --git a/src/runner/Synapse.Runner/Synapse.Runner.csproj b/src/runner/Synapse.Runner/Synapse.Runner.csproj index 31b8025a..dc1e6fe0 100644 --- a/src/runner/Synapse.Runner/Synapse.Runner.csproj +++ b/src/runner/Synapse.Runner/Synapse.Runner.csproj @@ -8,7 +8,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/runtime/Synapse.Runtime.Abstractions/Synapse.Runtime.Abstractions.csproj b/src/runtime/Synapse.Runtime.Abstractions/Synapse.Runtime.Abstractions.csproj index 2494eadc..8d479961 100644 --- a/src/runtime/Synapse.Runtime.Abstractions/Synapse.Runtime.Abstractions.csproj +++ b/src/runtime/Synapse.Runtime.Abstractions/Synapse.Runtime.Abstractions.csproj @@ -7,7 +7,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/runtime/Synapse.Runtime.Docker/Synapse.Runtime.Docker.csproj b/src/runtime/Synapse.Runtime.Docker/Synapse.Runtime.Docker.csproj index ec2a3dab..72020a79 100644 --- a/src/runtime/Synapse.Runtime.Docker/Synapse.Runtime.Docker.csproj +++ b/src/runtime/Synapse.Runtime.Docker/Synapse.Runtime.Docker.csproj @@ -7,7 +7,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/runtime/Synapse.Runtime.Kubernetes/Synapse.Runtime.Kubernetes.csproj b/src/runtime/Synapse.Runtime.Kubernetes/Synapse.Runtime.Kubernetes.csproj index 60e3a089..da1d7839 100644 --- a/src/runtime/Synapse.Runtime.Kubernetes/Synapse.Runtime.Kubernetes.csproj +++ b/src/runtime/Synapse.Runtime.Kubernetes/Synapse.Runtime.Kubernetes.csproj @@ -7,7 +7,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors diff --git a/src/runtime/Synapse.Runtime.Native/Synapse.Runtime.Native.csproj b/src/runtime/Synapse.Runtime.Native/Synapse.Runtime.Native.csproj index a35678e1..a97a7d6d 100644 --- a/src/runtime/Synapse.Runtime.Native/Synapse.Runtime.Native.csproj +++ b/src/runtime/Synapse.Runtime.Native/Synapse.Runtime.Native.csproj @@ -7,7 +7,7 @@ en True 1.0.0 - alpha5.8 + alpha5.11 $(VersionPrefix) $(VersionPrefix) The Synapse Authors