Implements support for the stream metadata (#53)

* Implements support for the stream metadata Part of SneaksAndData/arcane-framework#113 * Fix incorrect usage of incoming contract * Build fix * Use the stream metadata interface from Arcane.Framework v0.0.37
SneaksAndData · Oct 16, 2024 · ab3f435 · ab3f435
1 parent f978fb4
commit ab3f435
Show file tree

Hide file tree

Showing 4 changed files with 54 additions and 2 deletions.
diff --git a/.helm/templates/crd-sql-server-change-tracking.yaml b/.helm/templates/crd-sql-server-change-tracking.yaml
@@ -109,6 +109,22 @@ spec:
                   description: How long to wait before polling for next result set. Can be from 1 to 60 seconds.
                   minimum: 1
                   maximum: 60
+                streamMetadata:
+                  type: object
+                  optional: true
+                  default: null
+                  properties:
+                    partitions:
+                      type: array
+                      items:
+                        type: object
+                        properties:
+                          name:
+                            type: string
+                          fieldName:
+                            type: string
+                          fieldFormat:
+                            type: string
             status:
               type: object
               properties:

diff --git a/src/Arcane.Stream.SqlServerChangeTracking.csproj b/src/Arcane.Stream.SqlServerChangeTracking.csproj
@@ -5,7 +5,7 @@
         <TargetFramework>net8.0</TargetFramework>
     </PropertyGroup>
     <ItemGroup>
-        <PackageReference Include="Arcane.Framework" Version="0.0.36" />
+        <PackageReference Include="Arcane.Framework" Version="0.0.37" />
     </ItemGroup>
 
 </Project>
diff --git a/src/Models/SqlServerChangeTrackingStreamContext.cs b/src/Models/SqlServerChangeTrackingStreamContext.cs
@@ -1,7 +1,11 @@
 using System;
+using System.Linq;
 using System.Text.Json.Serialization;
+using Akka.Util;
 using Arcane.Framework.Configuration;
 using Arcane.Framework.Services.Base;
+using Arcane.Framework.Services.Models;
+using Arcane.Framework.Sinks.Models;
 
 namespace Arcane.Stream.SqlServerChangeTracking.Models;
 
@@ -69,6 +73,35 @@ public class SqlServerChangeTrackingStreamContext : IStreamContext, IStreamConte
     /// <inheritdoc cref="IStreamContext.StreamKind"/>
     public string StreamKind { get; private set; }
 
+    /// <summary>
+    /// Property to hold stream metadata received from the stream context.
+    /// </summary>
+    [JsonPropertyName("streamMetadata")]
+    public StreamMetadataDefinition StreamMetadataProperty { get; set; }
+
+    /// <inheritdoc cref="IStreamContext.StreamMetadata"/>
+    [JsonIgnore]
+    public Option<StreamMetadata> StreamMetadata
+    {
+        get
+        {
+            if (this.StreamMetadataProperty is null)
+            {
+                return Option<StreamMetadata>.None;
+            }
+
+            var partitions = this.StreamMetadataProperty
+                .Partitions
+                .Select(partition => new StreamPartition
+            {
+                Name = partition.Name,
+                FieldName = partition.FieldName,
+                FieldFormat = partition.FieldFormat
+            }).ToArray();
+            return new StreamMetadata(partitions);
+        }
+    }
+
     /// <inheritdoc cref="IStreamContextWriter.SetStreamId"/>
     public void SetStreamId(string streamId)
     {

diff --git a/src/Services/SqlServerChangeTrackgingGraphBuilder.cs b/src/Services/SqlServerChangeTrackgingGraphBuilder.cs
@@ -4,8 +4,10 @@
 using System.Threading.Tasks;
 using Akka.Streams;
 using Akka.Streams.Dsl;
+using Akka.Util;
 using Arcane.Framework.Contracts;
 using Arcane.Framework.Services.Base;
+using Arcane.Framework.Sinks.Models;
 using Arcane.Framework.Sinks.Parquet;
 using Arcane.Framework.Sources.Exceptions;
 using Arcane.Framework.Sources.SqlServer;
@@ -51,7 +53,8 @@ ILogger<SqlServerChangeTrackingGraphBuilder> logger
                 rowGroupsPerFile: context.GroupsPerFile,
                 createSchemaFile: true,
                 dataSinkPathSegment: context.IsBackfilling ? "backfill" : "data",
-                dropCompletionToken: context.IsBackfilling);
+                dropCompletionToken: context.IsBackfilling,
+                streamMetadata: context.StreamMetadata.GetOrElse(new StreamMetadata(Option<StreamPartition[]>.None)));
 
             return Source
                 .FromGraph(source)