Skip to content

Commit 48b4968

Browse files
cdavernascdavernas
authored
Merge pull request #478 from serverlessworkflow/feat-subscription-iterator
Implement streaming features for the `listen` task and for the `asyncapi` call
2 parents ac3af7e + 73b296a commit 48b4968

30 files changed

+583
-50
lines changed

‎src/api/Synapse.Api.Application/Synapse.Api.Application.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -44,7 +44,7 @@
4444
<ItemGroup>
4545
<PackageReference Include="IdentityServer4" Version="4.1.2" NoWarn="NU1902" />
4646
<PackageReference Include="IdentityServer4.Storage" Version="4.1.2" NoWarn="NU1902" />
47-
<PackageReference Include="Polly" Version="8.5.0" />
47+
<PackageReference Include="Polly" Version="8.5.1" />
4848
</ItemGroup>
4949

5050
<ItemGroup>

‎src/api/Synapse.Api.Client.Http/Synapse.Api.Client.Http.csproj

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -42,8 +42,8 @@
4242
</ItemGroup>
4343

4444
<ItemGroup>
45-
<PackageReference Include="Microsoft.AspNetCore.SignalR.Client" Version="9.0.0" />
46-
<PackageReference Include="ServerlessWorkflow.Sdk.IO" Version="1.0.0-alpha6.2" />
45+
<PackageReference Include="Microsoft.AspNetCore.SignalR.Client" Version="9.0.1" />
46+
<PackageReference Include="ServerlessWorkflow.Sdk.IO" Version="1.0.0-alpha6.3" />
4747
<PackageReference Include="System.Reactive" Version="6.0.1" />
4848
</ItemGroup>
4949

‎src/api/Synapse.Api.Server/Synapse.Api.Server.csproj

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -31,9 +31,9 @@
3131
</PropertyGroup>
3232

3333
<ItemGroup>
34-
<PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="9.0.0" />
35-
<PackageReference Include="Microsoft.AspNetCore.Authentication.OpenIdConnect" Version="9.0.0" />
36-
<PackageReference Include="Microsoft.AspNetCore.Components.WebAssembly.Server" Version="9.0.0" />
34+
<PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="9.0.1" />
35+
<PackageReference Include="Microsoft.AspNetCore.Authentication.OpenIdConnect" Version="9.0.1" />
36+
<PackageReference Include="Microsoft.AspNetCore.Components.WebAssembly.Server" Version="9.0.1" />
3737
<PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.21.0" />
3838
<PackageReference Include="Swashbuckle.AspNetCore.SwaggerUI" Version="7.2.0" />
3939
</ItemGroup>

‎src/cli/Synapse.Cli/Synapse.Cli.csproj

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -29,11 +29,11 @@
2929
</PropertyGroup>
3030

3131
<ItemGroup>
32-
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="9.0.0" />
33-
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="9.0.0" />
32+
<PackageReference Include="Microsoft.Extensions.DependencyInjection" Version="9.0.1" />
33+
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="9.0.1" />
3434
<PackageReference Include="moment.net" Version="1.3.4" />
3535
<PackageReference Include="NetEscapades.Configuration.Yaml" Version="3.1.0" />
36-
<PackageReference Include="ServerlessWorkflow.Sdk.IO" Version="1.0.0-alpha6.2" />
36+
<PackageReference Include="ServerlessWorkflow.Sdk.IO" Version="1.0.0-alpha6.3" />
3737
<PackageReference Include="Spectre.Console" Version="0.49.1" />
3838
<PackageReference Include="System.CommandLine.NamingConventionBinder" Version="2.0.0-beta4.22272.1" />
3939
</ItemGroup>

‎src/core/Synapse.Core.Infrastructure/Synapse.Core.Infrastructure.csproj

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -51,7 +51,7 @@
5151
<PackageReference Include="Neuroglia.Mediation" Version="4.18.1" />
5252
<PackageReference Include="Neuroglia.Plugins" Version="4.18.1" />
5353
<PackageReference Include="Neuroglia.Serialization.Xml" Version="4.18.1" />
54-
<PackageReference Include="ServerlessWorkflow.Sdk.IO" Version="1.0.0-alpha6.2" />
54+
<PackageReference Include="ServerlessWorkflow.Sdk.IO" Version="1.0.0-alpha6.3" />
5555
</ItemGroup>
5656

5757
<ItemGroup>

‎src/core/Synapse.Core/Resources/CorrelationContext.cs

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -37,13 +37,19 @@ public record CorrelationContext
3737
/// <summary>
3838
/// Gets a key/value mapping of the context's correlation keys
3939
/// </summary>
40-
[DataMember(Name = "keys", Order = 2), JsonPropertyName("keys"), JsonPropertyOrder(2), YamlMember(Alias = "keys", Order = 2)]
40+
[DataMember(Name = "keys", Order = 3), JsonPropertyName("keys"), JsonPropertyOrder(3), YamlMember(Alias = "keys", Order = 3)]
4141
public virtual EquatableDictionary<string, string> Keys { get; set; } = [];
4242

4343
/// <summary>
4444
/// Gets a key/value mapping of all correlated events, with the key being the index of the matched correlation filter
4545
/// </summary>
46-
[DataMember(Name = "events", Order = 3), JsonPropertyName("events"), JsonPropertyOrder(3), YamlMember(Alias = "events", Order = 3)]
46+
[DataMember(Name = "events", Order = 4), JsonPropertyName("events"), JsonPropertyOrder(4), YamlMember(Alias = "events", Order = 4)]
4747
public virtual EquatableDictionary<int, CloudEvent> Events { get; set; } = [];
4848

49+
/// <summary>
50+
/// Gets the offset that serves as the index of the event being processed by the consumer, if streaming has been enabled for the correlation associated with the context.
51+
/// </summary>
52+
[DataMember(Name = "offset", Order = 5), JsonPropertyName("offset"), JsonPropertyOrder(5), YamlMember(Alias = "offset", Order = 5)]
53+
public virtual uint? Offset { get; set; }
54+
4955
}

‎src/core/Synapse.Core/Resources/CorrelationSpec.cs

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,10 +46,16 @@ public record CorrelationSpec
4646
[DataMember(Name = "events", Order = 4), JsonPropertyName("events"), JsonPropertyOrder(4), YamlMember(Alias = "events", Order = 4)]
4747
public virtual EventConsumptionStrategyDefinition Events { get; set; } = null!;
4848

49+
/// <summary>
50+
/// Gets/sets a boolean indicating whether or not to stream events. When enabled, each correlated event is atomically published to the subscriber immediately rather than waiting for the entire correlation to complete
51+
/// </summary>
52+
[DataMember(Name = "stream", Order = 5), JsonPropertyName("stream"), JsonPropertyOrder(5), YamlMember(Alias = "stream", Order = 5)]
53+
public virtual bool Stream { get; set; }
54+
4955
/// <summary>
5056
/// Gets/sets an object used to configure the correlation's outcome
5157
/// </summary>
52-
[DataMember(Name = "outcome", Order = 5), JsonPropertyName("outcome"), JsonPropertyOrder(5), YamlMember(Alias = "outcome", Order = 5)]
58+
[DataMember(Name = "outcome", Order = 6), JsonPropertyName("outcome"), JsonPropertyOrder(6), YamlMember(Alias = "outcome", Order = 6)]
5359
public virtual CorrelationOutcomeDefinition Outcome { get; set; } = null!;
5460

5561
}

‎src/core/Synapse.Core/Synapse.Core.csproj

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -66,11 +66,11 @@
6666
<ItemGroup>
6767
<PackageReference Include="Apache.Avro" Version="1.12.0" />
6868
<PackageReference Include="Docker.DotNet" Version="3.125.15" />
69-
<PackageReference Include="KubernetesClient" Version="15.0.1" />
69+
<PackageReference Include="KubernetesClient" Version="16.0.1" />
7070
<PackageReference Include="Neuroglia.Data.Infrastructure.ResourceOriented" Version="4.18.1" />
7171
<PackageReference Include="Neuroglia.Eventing.CloudEvents" Version="4.18.1" />
7272
<PackageReference Include="Semver" Version="3.0.0" />
73-
<PackageReference Include="ServerlessWorkflow.Sdk" Version="1.0.0-alpha6.2" />
73+
<PackageReference Include="ServerlessWorkflow.Sdk" Version="1.0.0-alpha6.3" />
7474
</ItemGroup>
7575

7676
</Project>

‎src/correlator/Synapse.Correlator/Services/CorrelationHandler.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -334,7 +334,7 @@ protected virtual async Task CreateOrUpdateContextAsync(CorrelationContext conte
334334
{
335335
var index = updatedResource.Status.Contexts.IndexOf(existingContext);
336336
updatedResource.Status.Contexts.Remove(existingContext);
337-
if (!completed) updatedResource.Status.Contexts.Insert(index, context);
337+
updatedResource.Status.Contexts.Insert(index, context);
338338
}
339339
if (completed)
340340
{

‎src/correlator/Synapse.Correlator/Synapse.Correlator.csproj

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,8 +33,8 @@
3333
</PropertyGroup>
3434

3535
<ItemGroup>
36-
<PackageReference Include="Microsoft.Extensions.Configuration.KeyPerFile" Version="9.0.0" />
37-
<PackageReference Include="Microsoft.Extensions.Hosting" Version="9.0.0" />
36+
<PackageReference Include="Microsoft.Extensions.Configuration.KeyPerFile" Version="9.0.1" />
37+
<PackageReference Include="Microsoft.Extensions.Hosting" Version="9.0.1" />
3838
<PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.21.0" />
3939
<PackageReference Include="Neuroglia.Data.Expressions.JavaScript" Version="4.18.1" />
4040
<PackageReference Include="Neuroglia.Data.Expressions.JQ" Version="4.18.1" />

‎src/dashboard/Synapse.Dashboard.StateManagement/Synapse.Dashboard.StateManagement.csproj

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -9,9 +9,9 @@
99
</PropertyGroup>
1010

1111
<ItemGroup>
12-
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.0" />
13-
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="9.0.0" />
14-
<PackageReference Include="Microsoft.Extensions.Options" Version="9.0.0" />
12+
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="9.0.1" />
13+
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="9.0.1" />
14+
<PackageReference Include="Microsoft.Extensions.Options" Version="9.0.1" />
1515
</ItemGroup>
1616

1717
<ItemGroup>

‎src/dashboard/Synapse.Dashboard/Synapse.Dashboard.csproj

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,9 +13,9 @@
1313
<PackageReference Include="Blazor.Bootstrap" Version="3.2.0" />
1414
<PackageReference Include="BlazorMonaco" Version="3.3.0" />
1515
<PackageReference Include="IdentityModel" Version="7.0.0" />
16-
<PackageReference Include="Microsoft.AspNetCore.Components.WebAssembly" Version="9.0.0" />
17-
<PackageReference Include="Microsoft.AspNetCore.Components.WebAssembly.Authentication" Version="9.0.0" />
18-
<PackageReference Include="Microsoft.AspNetCore.Components.WebAssembly.DevServer" Version="9.0.0" PrivateAssets="all" />
16+
<PackageReference Include="Microsoft.AspNetCore.Components.WebAssembly" Version="9.0.1" />
17+
<PackageReference Include="Microsoft.AspNetCore.Components.WebAssembly.Authentication" Version="9.0.1" />
18+
<PackageReference Include="Microsoft.AspNetCore.Components.WebAssembly.DevServer" Version="9.0.1" PrivateAssets="all" />
1919
<PackageReference Include="moment.net" Version="1.3.4" />
2020
<PackageReference Include="Neuroglia.Blazor.Dagre" Version="4.18.1" />
2121
</ItemGroup>

‎src/operator/Synapse.Operator/Services/WorkflowInstanceController.cs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -156,7 +156,7 @@ protected override async Task OnResourceCreatedAsync(WorkflowInstance workflowIn
156156
}
157157
catch(Exception ex)
158158
{
159-
this.Logger.LogError("An error occured while handling the creation of workflow instance '{workflowInstance}': {ex}", workflowInstance.GetQualifiedName(), ex);
159+
this.Logger.LogError("An error occurred while handling the creation of workflow instance '{workflowInstance}': {ex}", workflowInstance.GetQualifiedName(), ex);
160160
}
161161
}
162162

‎src/operator/Synapse.Operator/Synapse.Operator.csproj

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,8 +51,8 @@
5151

5252
<ItemGroup>
5353
<PackageReference Include="Cronos" Version="0.9.0" />
54-
<PackageReference Include="Microsoft.Extensions.Configuration.KeyPerFile" Version="9.0.0" />
55-
<PackageReference Include="Microsoft.Extensions.Hosting" Version="9.0.0" />
54+
<PackageReference Include="Microsoft.Extensions.Configuration.KeyPerFile" Version="9.0.1" />
55+
<PackageReference Include="Microsoft.Extensions.Hosting" Version="9.0.1" />
5656
<PackageReference Include="Microsoft.VisualStudio.Azure.Containers.Tools.Targets" Version="1.21.0" />
5757
</ItemGroup>
5858

‎src/runner/Synapse.Runner/IStreamedCloudEvent.cs

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
// Copyright © 2024-Present The Synapse Authors
2+
//
3+
// Licensed under the Apache License, Version 2.0 (the "License"),
4+
// you may not use this file except in compliance with the License.
5+
// You may obtain a copy of the License at
6+
// http://www.apache.org/licenses/LICENSE-2.0
7+
//
8+
// Unless required by applicable law or agreed to in writing, software
9+
// distributed under the License is distributed on an "AS IS" BASIS,
10+
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11+
// See the License for the specific language governing permissions and
12+
// limitations under the License.
13+
14+
namespace Synapse.Runner;
15+
16+
/// <summary>
17+
/// Defines the fundamentals of an object used to wrap a streamed <see cref="CloudEvent"/>
18+
/// </summary>
19+
public interface IStreamedCloudEvent
20+
{
21+
22+
/// <summary>
23+
/// Gets the streamed <see cref="CloudEvent"/>
24+
/// </summary>
25+
CloudEvent Event { get; }
26+
27+
/// <summary>
28+
/// Gets the position of the <see cref="CloudEvent"/> within its originating stream
29+
/// </summary>
30+
uint Offset { get; }
31+
32+
/// <summary>
33+
/// Acknowledges that the <see cref="CloudEvent"/> has been successfully processed
34+
/// </summary>
35+
/// <param name="cancellationToken">A <see cref="CancellationToken"/></param>
36+
/// <returns>A new awaitable <see cref="Task"/></returns>
37+
Task AckAsync(CancellationToken cancellationToken = default);
38+
39+
}

‎src/runner/Synapse.Runner/Services/ConnectedWorkflowExecutionContext.cs

Lines changed: 116 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,8 @@
2020
using Synapse.Events.Tasks;
2121
using Synapse.Events.Workflows;
2222
using System.Net.Mime;
23+
using System.Reactive.Disposables;
24+
using System.Reactive.Threading.Tasks;
2325

2426
namespace Synapse.Runner.Services;
2527

@@ -304,11 +306,125 @@ public virtual async Task ResumeAsync(CancellationToken cancellationToken = defa
304306
this.Logger.LogInformation("The workflow's execution has been resumed.");
305307
}
306308

309+
/// <inheritdoc/>
310+
public virtual async Task<IObservable<IStreamedCloudEvent>> StreamAsync(ITaskExecutionContext task, CancellationToken cancellationToken = default)
311+
{
312+
ArgumentNullException.ThrowIfNull(task);
313+
if (task.Definition is not ListenTaskDefinition listenTask) throw new ArgumentException("The specified task's definition must be a 'listen' task", nameof(task));
314+
if (listenTask.Foreach == null) throw new ArgumentException($"Since the specified listen task doesn't use streaming, the {nameof(CorrelateAsync)} method must be used instead");
315+
if (this.Instance.Status?.Correlation?.Contexts?.TryGetValue(task.Instance.Reference.OriginalString, out var context) == true && context != null) return Observable.Empty<IStreamedCloudEvent>();
316+
var @namespace = task.Workflow.Instance.GetNamespace()!;
317+
var name = $"{task.Workflow.Instance.GetName()}.{task.Instance.Id}";
318+
Correlation? correlation = null;
319+
try { correlation = await this.Api.Correlations.GetAsync(name, @namespace, cancellationToken).ConfigureAwait(false); }
320+
catch { }
321+
if (correlation == null)
322+
{
323+
correlation = await this.Api.Correlations.CreateAsync(new()
324+
{
325+
Metadata = new()
326+
{
327+
Namespace = @namespace,
328+
Name = name,
329+
Labels = new Dictionary<string, string>()
330+
{
331+
{ SynapseDefaults.Resources.Labels.WorkflowInstance, this.Instance.GetQualifiedName() }
332+
}
333+
},
334+
Spec = new()
335+
{
336+
Source = new ResourceReference<WorkflowInstance>(task.Workflow.Instance.GetName(), task.Workflow.Instance.GetNamespace()),
337+
Lifetime = CorrelationLifetime.Ephemeral,
338+
Events = listenTask.Listen.To,
339+
Stream = true,
340+
Expressions = task.Workflow.Definition.Evaluate ?? new(),
341+
Outcome = new()
342+
{
343+
Correlate = new()
344+
{
345+
Instance = task.Workflow.Instance.GetQualifiedName(),
346+
Task = task.Instance.Reference.OriginalString
347+
}
348+
}
349+
}
350+
}, cancellationToken).ConfigureAwait(false);
351+
}
352+
var taskCompletionSource = new TaskCompletionSource<CorrelationContext>();
353+
var cancellationTokenRegistration = cancellationToken.Register(() => taskCompletionSource.TrySetCanceled());
354+
var correlationSubscription = this.Api.WorkflowInstances.MonitorAsync(this.Instance.GetName(), this.Instance.GetNamespace()!, cancellationToken)
355+
.ToObservable()
356+
.Where(e => e.Type == ResourceWatchEventType.Updated)
357+
.Select(e => e.Resource.Status?.Correlation?.Contexts)
358+
.Scan((Previous: (EquatableDictionary<string, CorrelationContext>?)null, Current: (EquatableDictionary<string, CorrelationContext>?)null), (accumulator, current) => (accumulator.Current ?? [], current))
359+
.Where(v => v.Current?.Count > v.Previous?.Count) //ensures we are not handling changes in a circular loop: if length of current is smaller than previous, it means a context has been processed
360+
.Subscribe(value =>
361+
{
362+
var patch = JsonPatchUtility.CreateJsonPatchFromDiff(value.Previous, value.Current);
363+
var patchOperation = patch.Operations.FirstOrDefault(o => o.Op == OperationType.Add && o.Path[0] == task.Instance.Reference.OriginalString);
364+
if (patchOperation == null) return;
365+
context = this.JsonSerializer.Deserialize<CorrelationContext>(patchOperation.Value!)!;
366+
taskCompletionSource.SetResult(context);
367+
});
368+
var endOfStream = false;
369+
var stopObservable = taskCompletionSource.Task.ToObservable();
370+
var stopSubscription = stopObservable.Take(1).Subscribe(_ => endOfStream = true);
371+
return Observable.Create<StreamedCloudEvent>(observer =>
372+
{
373+
var subscription = Observable.Using(
374+
() => new CompositeDisposable
375+
{
376+
cancellationTokenRegistration,
377+
correlationSubscription
378+
},
379+
disposable => this.Api.Correlations.MonitorAsync(correlation.GetName(), correlation.GetNamespace()!, cancellationToken)
380+
.ToObservable()
381+
.Where(e => e.Type == ResourceWatchEventType.Updated)
382+
.Select(e => e.Resource.Status?.Contexts?.FirstOrDefault())
383+
.Where(c => c != null)
384+
.SelectMany(c =>
385+
{
386+
var acknowledgedOffset = c!.Offset.HasValue ? (int)c.Offset.Value : 0;
387+
return c.Events.Values
388+
.Skip(acknowledgedOffset)
389+
.Select((evt, index) => new
390+
{
391+
ContextId = c.Id,
392+
Event = evt,
393+
Offset = (uint)(acknowledgedOffset + index + 1)
394+
});
395+
})
396+
.Distinct(e => e.Offset)
397+
.Select(e => new StreamedCloudEvent(e.Event, e.Offset, async (offset, token) =>
398+
{
399+
var original = await this.Api.Correlations.GetAsync(name, @namespace, token).ConfigureAwait(false);
400+
var updated = original.Clone()!;
401+
var context = updated.Status?.Contexts.FirstOrDefault(c => c.Id == e.ContextId);
402+
if (context == null)
403+
{
404+
this.Logger.LogError("Failed to find a context with the specified id '{contextId}' in correlation '{name}.{@namespace}'", e.ContextId, name, @namespace);
405+
throw new Exception($"Failed to find a context with the specified id '{e.ContextId}' in correlation '{name}.{@namespace}'");
406+
}
407+
context.Offset = offset;
408+
var patch = JsonPatchUtility.CreateJsonPatchFromDiff(original, updated);
409+
await this.Api.Correlations.PatchStatusAsync(name, @namespace, new Patch(PatchType.JsonPatch, patch), cancellationToken: token).ConfigureAwait(false);
410+
})))
411+
.Subscribe(e =>
412+
{
413+
observer.OnNext(e);
414+
if (endOfStream) observer.OnCompleted();
415+
},
416+
ex => observer.OnError(ex),
417+
() => observer.OnCompleted());
418+
return new CompositeDisposable(subscription, stopSubscription);
419+
});
420+
}
421+
307422
/// <inheritdoc/>
308423
public virtual async Task<CorrelationContext> CorrelateAsync(ITaskExecutionContext task, CancellationToken cancellationToken = default)
309424
{
310425
ArgumentNullException.ThrowIfNull(task);
311426
if (task.Definition is not ListenTaskDefinition listenTask) throw new ArgumentException("The specified task's definition must be a 'listen' task", nameof(task));
427+
if (listenTask.Foreach == null) throw new ArgumentException($"Since the specified listen task uses streaming, the {nameof(StreamAsync)} method must be used instead");
312428
if (this.Instance.Status?.Correlation?.Contexts?.TryGetValue(task.Instance.Reference.OriginalString, out var context) == true && context != null) return context;
313429
var @namespace = task.Workflow.Instance.GetNamespace()!;
314430
var name = $"{task.Workflow.Instance.GetName()}.{task.Instance.Id}";

0 commit comments

Comments
 (0)