-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #5 from RobThree/mssql
SQL Server implementation
- Loading branch information
Showing
28 changed files
with
711 additions
and
33 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,48 @@ | ||
| #!/usr/bin/env bash | ||
|
|
||
| # Start SQL Server | ||
| /opt/mssql/bin/sqlservr & | ||
|
|
||
| # Wait for SQL to start | ||
| TRIES=60 | ||
| DBSTATUS=1 | ||
| ERRCODE=1 | ||
| i=0 | ||
|
|
||
| echo "Waiting for SQL Server to start" | ||
| while [[ $DBSTATUS -ne 0 ]] && [[ $i -lt $TRIES ]]; do | ||
| i=$((i+1)) | ||
| DBSTATUS=$(/opt/mssql-tools/bin/sqlcmd -h -1 -t 1 -U sa -P $MSSQL_SA_PASSWORD -Q "SET NOCOUNT ON; Select COALESCE(SUM(state), 0) from sys.databases") || DBSTATUS=1 | ||
| if [ $DBSTATUS -ne 0 ]; then | ||
| sleep 1s | ||
| fi | ||
| done | ||
|
|
||
| sleep 5s | ||
| if [ $DBSTATUS -ne 0 ]; then | ||
| echo "SQL Server took more than $TRIES seconds to start up or one or more databases are not in an ONLINE state" | ||
| exit 1 | ||
| fi | ||
|
|
||
| mssql=( /opt/mssql-tools/bin/sqlcmd -S localhost -U sa -P "$MSSQL_SA_PASSWORD" -d master -i ) | ||
|
|
||
| find /docker-entrypoint-initdb.d -mindepth 2 -type f | sort | while read f; do | ||
| case "$f" in | ||
| *.sh) | ||
| if [ -x "$f" ]; then | ||
| echo "$0: running $f" | ||
| "$f" | ||
| else | ||
| echo "$0: sourcing $f" | ||
| . "$f" | ||
| fi | ||
| ;; | ||
| *.sql) echo "$0: running $f"; "${mssql[@]}" "$f"; echo ;; | ||
| *.sql.gz) echo "$0: running $f"; gunzip -c "$f" | "${mssql[@]}"; echo ;; | ||
| *) echo "$0: ignoring $f" ;; | ||
| esac | ||
| echo | ||
| done | ||
|
|
||
| echo "SQL Server is running" | ||
| sleep infinity |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
17 changes: 17 additions & 0 deletions
17
src/Contrib.KafkaFlow.Outbox.SqlServer/ConfigurationBuilderExtensions.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| using KafkaFlow; | ||
| using KafkaFlow.SqlServer; | ||
| using Microsoft.Extensions.DependencyInjection; | ||
|
|
||
| namespace KafkaFlow.Outbox.SqlServer; | ||
|
|
||
| public static class ConfigurationBuilderExtensions | ||
| { | ||
| public static IServiceCollection AddSqlServerOutboxBackend(this IServiceCollection services) => | ||
| services.AddSingleton<IOutboxBackend, SqlServerOutboxBackend>(); | ||
|
|
||
| public static IServiceCollection AddSqlServerOutboxBackend(this IServiceCollection services, string connectionString) | ||
| { | ||
| services.ConfigureSqlServerBackend(options => options.ConnectionString = connectionString); | ||
| return AddSqlServerOutboxBackend(services); | ||
| } | ||
| } |
26 changes: 26 additions & 0 deletions
26
src/Contrib.KafkaFlow.Outbox.SqlServer/Contrib.KafkaFlow.Outbox.SqlServer.csproj
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,26 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <TargetFramework>net8.0</TargetFramework> | ||
| <ImplicitUsings>enable</ImplicitUsings> | ||
| <Nullable>enable</Nullable> | ||
| <LangVersion>latest</LangVersion> | ||
| <EnableNETAnalyzers>true</EnableNETAnalyzers> | ||
| <AnalysisLevel>latest</AnalysisLevel> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\Contrib.KafkaFlow.SqlServer\Contrib.KafkaFlow.SqlServer.csproj" /> | ||
| <ProjectReference Include="..\Contrib.KafkaFlow.Outbox\Contrib.KafkaFlow.Outbox.csproj" /> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup> | ||
| <PackageReference Include="Dapper" Version="2.1.35" /> | ||
| <PackageReference Include="Microsoft.Extensions.Configuration.Abstractions" Version="8.0.0" /> | ||
| <PackageReference Include="Microsoft.Extensions.Configuration.Binder" Version="8.0.1" /> | ||
| <PackageReference Include="Microsoft.Extensions.Hosting" Version="8.0.0" /> | ||
| <PackageReference Include="Microsoft.Extensions.Options.DataAnnotations" Version="8.0.0" /> | ||
| <PackageReference Include="System.Data.SqlClient" Version="4.8.6" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
95 changes: 95 additions & 0 deletions
95
src/Contrib.KafkaFlow.Outbox.SqlServer/SqlServerOutboxBackend.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| using Confluent.Kafka; | ||
| using Dapper; | ||
| using KafkaFlow.SqlServer; | ||
| using Microsoft.Extensions.Options; | ||
| using System.Data.SqlClient; | ||
| using System.Text.Json; | ||
|
|
||
| namespace KafkaFlow.Outbox.SqlServer; | ||
|
|
||
| public class SqlServerOutboxBackend : IOutboxBackend | ||
| { | ||
| private readonly SqlServerBackendOptions _options; | ||
|
|
||
| public SqlServerOutboxBackend(IOptions<SqlServerBackendOptions> options) | ||
| => _options = options?.Value ?? throw new ArgumentNullException(nameof(options)); | ||
|
|
||
| public async ValueTask Store(TopicPartition topicPartition, Message<byte[], byte[]> message, CancellationToken token = default) | ||
| { | ||
| var sql = """ | ||
| INSERT INTO [outbox].[outbox] ([topic_name], [partition], [message_key], [message_headers], [message_body]) | ||
| VALUES (@topic_name, @partition, @message_key, @message_headers, @message_body); | ||
| """; | ||
|
|
||
| using var conn = new SqlConnection(_options.ConnectionString); | ||
|
|
||
| var rawHeaders = | ||
| message.Headers == null | ||
| ? null | ||
| : JsonSerializer.Serialize(message.Headers.ToDictionary(x => x.Key, x => x.GetValueBytes())); | ||
|
|
||
| var res = await conn.ExecuteAsync(sql, new | ||
| { | ||
| topic_name = topicPartition.Topic, | ||
| partition = topicPartition.Partition.IsSpecial ? null : (int?)topicPartition.Partition.Value, | ||
| message_key = message.Key, | ||
| message_headers = rawHeaders, | ||
| message_body = message.Value | ||
| }).ConfigureAwait(false); | ||
|
|
||
| } | ||
|
|
||
| public async ValueTask<OutboxRecord[]> Read(int batchSize, CancellationToken token = default) | ||
| { | ||
| var sql = """ | ||
| DELETE FROM [outbox].[outbox] | ||
| OUTPUT DELETED.* | ||
| WHERE | ||
| [sequence_id] IN ( | ||
| SELECT TOP (@batch_size) [sequence_id] FROM [outbox].[outbox] | ||
| ORDER BY [sequence_id] | ||
| ); | ||
| """; | ||
| using var conn = new SqlConnection(_options.ConnectionString); | ||
| var result = await conn.QueryAsync<OutboxTableRow>(sql, new { batch_size = batchSize }); | ||
|
|
||
| return result?.Select(ToOutboxRecord).ToArray() ?? Array.Empty<OutboxRecord>(); | ||
| } | ||
|
|
||
| private static OutboxRecord ToOutboxRecord(OutboxTableRow row) | ||
| { | ||
| var partition = row.partition.HasValue ? new Partition(row.partition.Value) : Partition.Any; | ||
| var topicPartition = new TopicPartition(row.topic_name, partition); | ||
|
|
||
| var storedHeaders = | ||
| row.message_headers == null | ||
| ? null | ||
| : JsonSerializer.Deserialize<Dictionary<string, byte[]>>(row.message_headers); | ||
|
|
||
| var headers = | ||
| storedHeaders?.Aggregate(new Headers(), (h, x) => | ||
| { | ||
| h.Add(x.Key, x.Value); | ||
| return h; | ||
| }); | ||
|
|
||
| var msg = new Message<byte[], byte[]> | ||
| { | ||
| Key = row.message_key!, | ||
| Value = row.message_body!, | ||
| Headers = headers | ||
| }; | ||
|
|
||
| return new OutboxRecord(topicPartition, msg); | ||
| } | ||
| } | ||
|
|
||
| internal sealed class OutboxTableRow | ||
| { | ||
| public long sequence_id { get; set; } | ||
| public string topic_name { get; set; } = null!; | ||
| public int? partition { get; set; } | ||
| public byte[]? message_key { get; set; } | ||
| public string? message_headers { get; set; } | ||
| public byte[]? message_body { get; set; } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,5 @@ | ||
| IF NOT EXISTS(SELECT * FROM sys.schemas WHERE name = 'outbox') | ||
| BEGIN | ||
| EXEC ('CREATE SCHEMA [outbox] AUTHORIZATION [dbo]') | ||
| END | ||
| GO |
18 changes: 18 additions & 0 deletions
18
src/Contrib.KafkaFlow.Outbox.SqlServer/schema/0002.Table.sql
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| /* Table */ | ||
| IF OBJECT_ID(N'[outbox].[outbox]', N'U') IS NULL | ||
| BEGIN | ||
| CREATE TABLE [outbox].[outbox]( | ||
| [sequence_id] [bigint] IDENTITY(1,1) NOT NULL, | ||
| [topic_name] [nvarchar](255) NOT NULL, | ||
| [partition] [int] NULL, | ||
| [message_key] [varbinary](max) NULL, | ||
| [message_headers] [nvarchar](max) NULL, | ||
| [message_body] [varbinary](max) NULL, | ||
| [date_added_utc] [datetime2] NOT NULL DEFAULT(SYSUTCDATETIME()), | ||
| [rowversion] [int] NOT NULL DEFAULT(1), | ||
| CONSTRAINT [PK_outbox] PRIMARY KEY CLUSTERED ([sequence_id] ASC), | ||
| CONSTRAINT [CK_headers_not_blank_or_empty] CHECK ((TRIM([message_headers])<>N'')), | ||
| CONSTRAINT [CK_topic_name_not_blank_or_empty] CHECK ((TRIM([topic_name])<>N'')) | ||
| ) | ||
| END | ||
| GO |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
18 changes: 18 additions & 0 deletions
18
src/Contrib.KafkaFlow.ProcessManagers.SqlServer/ConfigurationBuilderExtensions.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| using KafkaFlow; | ||
| using KafkaFlow.SqlServer; | ||
| using Microsoft.Extensions.DependencyInjection; | ||
|
|
||
| namespace KafkaFlow.ProcessManagers.SqlServer; | ||
|
|
||
| public static class ConfigurationBuilderExtensions | ||
| { | ||
| public static IServiceCollection AddSqlServerProcessManagerState(this IServiceCollection services) => | ||
| services.AddSingleton<IProcessStateStore, SqlServerProcessManagersStore>(); | ||
|
|
||
|
|
||
| public static IServiceCollection AddSqlServerProcessManagerState(this IServiceCollection services, string connectionString) | ||
| { | ||
| services.ConfigureSqlServerBackend(options => options.ConnectionString = connectionString); | ||
| return AddSqlServerProcessManagerState(services); | ||
| } | ||
| } |
Oops, something went wrong.