Skip to content

Commit 5cf98c0

Browse files
cdavernascdavernas
authored
Merge pull request #430 from serverlessworkflow/feat-local-function-catalog
Added a new CustomFunction namespaced resource
2 parents c173756 + 7d27f01 commit 5cf98c0

File tree

58 files changed

+1888
-66
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

58 files changed

+1888
-66
lines changed

‎src/api/Synapse.Api.Application/Services/CloudEventPublisher.cs‎

Lines changed: 13 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -84,14 +84,21 @@ public class CloudEventPublisher(ILogger<CloudEventPublisher> logger, IConnectio
8484
protected bool SupportsStreaming { get; private set; }
8585

8686
/// <inheritdoc/>
87-
public virtual Task StartAsync(CancellationToken cancellationToken)
87+
public virtual async Task StartAsync(CancellationToken cancellationToken)
8888
{
8989
if (options.Value.CloudEvents.Endpoint == null) logger.LogWarning("No endpoint configured for cloud events. Events will not be published.");
9090
else _ = this.PublishEnqueuedEventsAsync();
9191
var version = ((string)(this.Database.Execute("INFO", "server"))!).Split('\n').FirstOrDefault(line => line.StartsWith("redis_version:"))?[14..]?.Trim() ?? "undetermined";
9292
try
9393
{
94-
this.Database.StreamInfo(SynapseDefaults.CloudEvents.Bus.StreamName);
94+
try
95+
{
96+
await this.Database.StreamInfoAsync(SynapseDefaults.CloudEvents.Bus.StreamName).ConfigureAwait(false);
97+
}
98+
catch (RedisServerException ex) when (ex.Message.StartsWith("ERR no such key"))
99+
{
100+
this.Logger.LogWarning("The cloud event stream is currently unavailable, but it should be created when cloud events are published or when correlators are activated");
101+
}
95102
this.SupportsStreaming = true;
96103
this.Logger.LogInformation("Redis server version '{version}' supports streaming commands. Streaming feature is enabled", version);
97104
}
@@ -100,7 +107,10 @@ public virtual Task StartAsync(CancellationToken cancellationToken)
100107
this.SupportsStreaming = false;
101108
this.Logger.LogInformation("Redis server version '{version}' does not support streaming commands. Streaming feature is emulated using lists", version);
102109
}
103-
return Task.CompletedTask;
110+
catch(Exception ex)
111+
{
112+
this.Logger.LogWarning("An error occurred while starting the cloud event publisher, possibly affecting the server's ability to publish cloud events to correlators: {ex}", ex);
113+
}
104114
}
105115

106116
/// <inheritdoc/>

‎src/api/Synapse.Api.Application/Services/WorkflowDatabaseInitializer.cs‎ renamed to ‎src/api/Synapse.Api.Application/Services/DatabaseProvisioner .cs‎

Lines changed: 132 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
using Microsoft.Extensions.Hosting;
1616
using Microsoft.Extensions.Logging;
1717
using Microsoft.Extensions.Options;
18+
using Neuroglia.Serialization;
1819
using ServerlessWorkflow.Sdk.IO;
1920
using ServerlessWorkflow.Sdk.Models;
2021
using Synapse.Api.Application.Configuration;
@@ -28,9 +29,11 @@ namespace Synapse.Api.Application.Services;
2829
/// </summary>
2930
/// <param name="serviceProvider">The current <see cref="IServiceProvider"/></param>
3031
/// <param name="logger">The service used to perform logging</param>
32+
/// <param name="jsonSerializer">The service used to serialize/deserialize data to/from JSON</param>
33+
/// <param name="yamlSerializer">The service used to serialize/deserialize data to/from YAML</param>
3134
/// <param name="workflowDefinitionReader">The service used to read <see cref="WorkflowDefinition"/>s</param>
3235
/// <param name="options">The service used to access the current <see cref="ApiServerOptions"/></param>
33-
public class WorkflowDatabaseInitializer(IServiceProvider serviceProvider, ILogger<WorkflowDatabaseInitializer> logger, IWorkflowDefinitionReader workflowDefinitionReader, IOptions<ApiServerOptions> options)
36+
public class WorkflowDatabaseInitializer(IServiceProvider serviceProvider, ILogger<WorkflowDatabaseInitializer> logger, IJsonSerializer jsonSerializer, IYamlSerializer yamlSerializer, IWorkflowDefinitionReader workflowDefinitionReader, IOptions<ApiServerOptions> options)
3437
: IHostedService
3538
{
3639

@@ -44,6 +47,16 @@ public class WorkflowDatabaseInitializer(IServiceProvider serviceProvider, ILogg
4447
/// </summary>
4548
protected ILogger Logger { get; } = logger;
4649

50+
/// <summary>
51+
/// Gets the service used to serialize/deserialize data to/from JSON
52+
/// </summary>
53+
protected IJsonSerializer JsonSerializer { get; } = jsonSerializer;
54+
55+
/// <summary>
56+
/// Gets the service used to serialize/deserialize data to/from YAML
57+
/// </summary>
58+
protected IYamlSerializer YamlSerializer { get; } = yamlSerializer;
59+
4760
/// <summary>
4861
/// Gets the service used to read <see cref="WorkflowDefinition"/>s
4962
/// </summary>
@@ -80,15 +93,80 @@ public virtual async Task StartAsync(CancellationToken cancellationToken)
8093
this.Logger.LogWarning("The directory '{directory}' does not exist or cannot be found. Skipping static resource import", directory.FullName);
8194
return;
8295
}
83-
this.Logger.LogInformation("Starting importing static resources from directory '{directory}'...", directory.FullName);
96+
await this.ProvisionNamespacesAsync(resources, cancellationToken).ConfigureAwait(false);
97+
await this.ProvisionWorkflowsAsync(resources, cancellationToken).ConfigureAwait(false);
98+
await this.ProvisionFunctionsAsync(resources, cancellationToken).ConfigureAwait(false);
99+
}
100+
101+
/// <summary>
102+
/// Provisions namespaces from statis resource files
103+
/// </summary>
104+
/// <param name="resources">The <see cref="IResourceRepository"/> used to manage <see cref="IResource"/>s</param>
105+
/// <param name="cancellationToken">A <see cref="CancellationToken"/></param>
106+
/// <returns>A new awaitable <see cref="Task"/></returns>
107+
protected virtual async Task ProvisionNamespacesAsync(IResourceRepository resources, CancellationToken cancellationToken)
108+
{
109+
var stopwatch = new Stopwatch();
110+
var directory = new DirectoryInfo(Path.Combine(this.Options.Seeding.Directory, "namespaces"));
111+
if (!directory.Exists) return;
112+
this.Logger.LogInformation("Starting importing namespaces from directory '{directory}'...", directory.FullName);
84113
var files = directory.GetFiles(this.Options.Seeding.FilePattern, SearchOption.AllDirectories).Where(f => f.FullName.EndsWith(".json", StringComparison.OrdinalIgnoreCase) || f.FullName.EndsWith(".yml", StringComparison.OrdinalIgnoreCase) || f.FullName.EndsWith(".yaml", StringComparison.OrdinalIgnoreCase));
85114
if (!files.Any())
86115
{
87-
this.Logger.LogWarning("No static resource files matching search pattern '{pattern}' found in directory '{directory}'. Skipping import.", this.Options.Seeding.FilePattern, directory.FullName);
116+
this.Logger.LogWarning("No namespace static resource files matching search pattern '{pattern}' found in directory '{directory}'. Skipping import.", this.Options.Seeding.FilePattern, directory.FullName);
88117
return;
89118
}
119+
stopwatch.Restart();
90120
var count = 0;
121+
foreach (var file in files)
122+
{
123+
try
124+
{
125+
var extension = file.FullName.Split('.', StringSplitOptions.RemoveEmptyEntries).LastOrDefault();
126+
var serializer = extension?.ToLowerInvariant() switch
127+
{
128+
"json" => (ITextSerializer)this.JsonSerializer,
129+
"yml" or "yaml" => this.YamlSerializer,
130+
_ => throw new NotSupportedException($"The specified extension '{extension}' is not supported for static resource files")
131+
};
132+
using var stream = file.OpenRead();
133+
using var streamReader = new StreamReader(stream);
134+
var text = await streamReader.ReadToEndAsync(cancellationToken).ConfigureAwait(false);
135+
var ns = serializer.Deserialize<NamespaceDefinition>(text)!;
136+
await resources.AddAsync(ns, false, cancellationToken).ConfigureAwait(false);
137+
this.Logger.LogInformation("Successfully imported namespace '{namespace}' from file '{file}'", $"{ns.Metadata.Name}", file.FullName);
138+
count++;
139+
}
140+
catch (Exception ex)
141+
{
142+
this.Logger.LogError("An error occurred while reading a namespace from file '{file}': {ex}", file.FullName, ex);
143+
continue;
144+
}
145+
}
146+
stopwatch.Stop();
147+
this.Logger.LogInformation("Completed importing {count} namespaces in {ms} milliseconds", count, stopwatch.Elapsed.TotalMilliseconds);
148+
}
149+
150+
/// <summary>
151+
/// Provisions workflows from statis resource files
152+
/// </summary>
153+
/// <param name="resources">The <see cref="IResourceRepository"/> used to manage <see cref="IResource"/>s</param>
154+
/// <param name="cancellationToken">A <see cref="CancellationToken"/></param>
155+
/// <returns>A new awaitable <see cref="Task"/></returns>
156+
protected virtual async Task ProvisionWorkflowsAsync(IResourceRepository resources, CancellationToken cancellationToken)
157+
{
158+
var stopwatch = new Stopwatch();
159+
var directory = new DirectoryInfo(Path.Combine(this.Options.Seeding.Directory, "workflows"));
160+
if (!directory.Exists) return;
161+
this.Logger.LogInformation("Starting importing workflows from directory '{directory}'...", directory.FullName);
162+
var files = directory.GetFiles(this.Options.Seeding.FilePattern, SearchOption.AllDirectories).Where(f => f.FullName.EndsWith(".json", StringComparison.OrdinalIgnoreCase) || f.FullName.EndsWith(".yml", StringComparison.OrdinalIgnoreCase) || f.FullName.EndsWith(".yaml", StringComparison.OrdinalIgnoreCase));
163+
if (!files.Any())
164+
{
165+
this.Logger.LogWarning("No workflow static resource files matching search pattern '{pattern}' found in directory '{directory}'. Skipping import.", this.Options.Seeding.FilePattern, directory.FullName);
166+
return;
167+
}
91168
stopwatch.Restart();
169+
var count = 0;
92170
foreach (var file in files)
93171
{
94172
try
@@ -110,11 +188,6 @@ public virtual async Task StartAsync(CancellationToken cancellationToken)
110188
Versions = [workflowDefinition]
111189
}
112190
};
113-
if (await resources.GetAsync<Namespace>(workflow.GetNamespace()!, cancellationToken: cancellationToken).ConfigureAwait(false) == null)
114-
{
115-
await resources.AddAsync(new Namespace() { Metadata = new() { Name = workflow.GetNamespace()! } }, false, cancellationToken).ConfigureAwait(false);
116-
this.Logger.LogInformation("Successfully created namespace '{namespace}'", workflow.GetNamespace());
117-
}
118191
await resources.AddAsync(workflow, false, cancellationToken).ConfigureAwait(false);
119192
}
120193
else
@@ -138,14 +211,63 @@ public virtual async Task StartAsync(CancellationToken cancellationToken)
138211
this.Logger.LogInformation("Successfully imported workflow '{workflow}' from file '{file}'", $"{workflowDefinition.Document.Name}.{workflowDefinition.Document.Namespace}:{workflowDefinition.Document.Version}", file.FullName);
139212
count++;
140213
}
141-
catch(Exception ex)
214+
catch (Exception ex)
142215
{
143216
this.Logger.LogError("An error occurred while reading a workflow definition from file '{file}': {ex}", file.FullName, ex);
144217
continue;
145218
}
146219
}
147220
stopwatch.Stop();
148-
this.Logger.LogInformation("Completed importing {count} static resources in {ms} milliseconds", count, stopwatch.Elapsed.TotalMilliseconds);
221+
this.Logger.LogInformation("Completed importing {count} workflows in {ms} milliseconds", count, stopwatch.Elapsed.TotalMilliseconds);
222+
}
223+
224+
/// <summary>
225+
/// Provisions functions from statis resource files
226+
/// </summary>
227+
/// <param name="resources">The <see cref="IResourceRepository"/> used to manage <see cref="IResource"/>s</param>
228+
/// <param name="cancellationToken">A <see cref="CancellationToken"/></param>
229+
/// <returns>A new awaitable <see cref="Task"/></returns>
230+
protected virtual async Task ProvisionFunctionsAsync(IResourceRepository resources, CancellationToken cancellationToken)
231+
{
232+
var stopwatch = new Stopwatch();
233+
var directory = new DirectoryInfo(Path.Combine(this.Options.Seeding.Directory, "functions"));
234+
if (!directory.Exists) return;
235+
this.Logger.LogInformation("Starting importing custom functions from directory '{directory}'...", directory.FullName);
236+
var files = directory.GetFiles(this.Options.Seeding.FilePattern, SearchOption.AllDirectories).Where(f => f.FullName.EndsWith(".json", StringComparison.OrdinalIgnoreCase) || f.FullName.EndsWith(".yml", StringComparison.OrdinalIgnoreCase) || f.FullName.EndsWith(".yaml", StringComparison.OrdinalIgnoreCase));
237+
if (!files.Any())
238+
{
239+
this.Logger.LogWarning("No custom function static resource files matching search pattern '{pattern}' found in directory '{directory}'. Skipping import.", this.Options.Seeding.FilePattern, directory.FullName);
240+
return;
241+
}
242+
stopwatch.Restart();
243+
var count = 0;
244+
foreach (var file in files)
245+
{
246+
try
247+
{
248+
var extension = file.FullName.Split('.', StringSplitOptions.RemoveEmptyEntries).LastOrDefault();
249+
var serializer = extension?.ToLowerInvariant() switch
250+
{
251+
"json" => (ITextSerializer)this.JsonSerializer,
252+
"yml" or "yaml" => this.YamlSerializer,
253+
_ => throw new NotSupportedException($"The specified extension '{extension}' is not supported for static resource files")
254+
};
255+
using var stream = file.OpenRead();
256+
using var streamReader = new StreamReader(stream);
257+
var text = await streamReader.ReadToEndAsync(cancellationToken).ConfigureAwait(false);
258+
var func = serializer.Deserialize<CustomFunction>(text)!;
259+
await resources.AddAsync(func, false, cancellationToken).ConfigureAwait(false);
260+
this.Logger.LogInformation("Successfully imported custom function '{customFunction}' from file '{file}'", func.GetQualifiedName(), file.FullName);
261+
count++;
262+
}
263+
catch (Exception ex)
264+
{
265+
this.Logger.LogError("An error occurred while reading a custom function from file '{file}': {ex}", file.FullName, ex);
266+
continue;
267+
}
268+
}
269+
stopwatch.Stop();
270+
this.Logger.LogInformation("Completed importing {count} custom functions in {ms} milliseconds", count, stopwatch.Elapsed.TotalMilliseconds);
149271
}
150272

151273
/// <inheritdoc/>

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
<NeutralLanguage>en</NeutralLanguage>
88
<GenerateDocumentationFile>True</GenerateDocumentationFile>
99
<VersionPrefix>1.0.0</VersionPrefix>
10-
<VersionSuffix>alpha3.3</VersionSuffix>
10+
<VersionSuffix>alpha4</VersionSuffix>
1111
<AssemblyVersion>$(VersionPrefix)</AssemblyVersion>
1212
<FileVersion>$(VersionPrefix)</FileVersion>
1313
<Authors>The Synapse Authors</Authors>

‎src/api/Synapse.Api.Client.Core/Services/ISynapseApiClient.cs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,11 @@ public interface ISynapseApiClient
3131
/// </summary>
3232
INamespacedResourceApiClient<Correlator> Correlators { get; }
3333

34+
/// <summary>
35+
/// Gets the Synapse API used to manage <see cref="CustomFunction"/>s
36+
/// </summary>
37+
INamespacedResourceApiClient<CustomFunction> CustomFunctions { get; }
38+
3439
/// <summary>
3540
/// Gets the Synapse API used to manage <see cref="Document"/>s
3641
/// </summary>

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
<NeutralLanguage>en</NeutralLanguage>
88
<GenerateDocumentationFile>True</GenerateDocumentationFile>
99
<VersionPrefix>1.0.0</VersionPrefix>
10-
<VersionSuffix>alpha3.3</VersionSuffix>
10+
<VersionSuffix>alpha4</VersionSuffix>
1111
<AssemblyVersion>$(VersionPrefix)</AssemblyVersion>
1212
<FileVersion>$(VersionPrefix)</FileVersion>
1313
<Authors>The Synapse Authors</Authors>

‎src/api/Synapse.Api.Client.Http/Services/SynapseHttpApiClient.cs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -74,6 +74,9 @@ public SynapseHttpApiClient(IServiceProvider serviceProvider, ILoggerFactory log
7474
/// <inheritdoc/>
7575
public INamespacedResourceApiClient<Correlator> Correlators { get; private set; } = null!;
7676

77+
/// <inheritdoc/>
78+
public INamespacedResourceApiClient<CustomFunction> CustomFunctions { get; private set; } = null!;
79+
7780
/// <inheritdoc/>
7881
public IDocumentApiClient Documents { get; }
7982

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
<NeutralLanguage>en</NeutralLanguage>
88
<GenerateDocumentationFile>True</GenerateDocumentationFile>
99
<VersionPrefix>1.0.0</VersionPrefix>
10-
<VersionSuffix>alpha3.3</VersionSuffix>
10+
<VersionSuffix>alpha4</VersionSuffix>
1111
<AssemblyVersion>$(VersionPrefix)</AssemblyVersion>
1212
<FileVersion>$(VersionPrefix)</FileVersion>
1313
<Authors>The Synapse Authors</Authors>

‎src/api/Synapse.Api.Http/Controllers/CustomFunctionsController.cs‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
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.Api.Http.Controllers;
15+
16+
/// <summary>
17+
/// Represents the <see cref="NamespacedResourceController{TResource}"/> used to manage <see cref="CustomFunction"/>s
18+
/// </summary>
19+
/// <param name="mediator">The service used to mediate calls</param>
20+
/// <param name="jsonSerializer">The service used to serialize/deserialize objects to/from JSON</param>
21+
[Route("api/v1/custom-functions")]
22+
public class CustomFunctionsController(IMediator mediator, IJsonSerializer jsonSerializer)
23+
: NamespacedResourceController<CustomFunction>(mediator, jsonSerializer)
24+
{
25+
26+
27+
28+
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@
88
<OutputType>Library</OutputType>
99
<GenerateDocumentationFile>True</GenerateDocumentationFile>
1010
<VersionPrefix>1.0.0</VersionPrefix>
11-
<VersionSuffix>alpha3.3</VersionSuffix>
11+
<VersionSuffix>alpha4</VersionSuffix>
1212
<AssemblyVersion>$(VersionPrefix)</AssemblyVersion>
1313
<FileVersion>$(VersionPrefix)</FileVersion>
1414
<Authors>The Synapse Authors</Authors>

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
<NeutralLanguage>en</NeutralLanguage>
88
<GenerateDocumentationFile>True</GenerateDocumentationFile>
99
<VersionPrefix>1.0.0</VersionPrefix>
10-
<VersionSuffix>alpha3.3</VersionSuffix>
10+
<VersionSuffix>alpha4</VersionSuffix>
1111
<AssemblyVersion>$(VersionPrefix)</AssemblyVersion>
1212
<FileVersion>$(VersionPrefix)</FileVersion>
1313
<Authors>The Synapse Authors</Authors>

0 commit comments

Comments
 (0)