Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
16 changed files
with
498 additions
and
12 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -25,6 +25,7 @@ packages.config | |
| ## JetBrains Rider | ||
| .idea/ | ||
| *.sln.iml | ||
| *.DotSettings.user | ||
|
|
||
| ## CodeRush | ||
| .cr/ | ||
|
|
||
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
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,13 @@ | ||
| namespace Propulsion.MessageDb | ||
|
|
||
| open FSharp.UMX | ||
| open Npgsql | ||
|
|
||
| module internal FeedSourceId = | ||
| let wellKnownId : Propulsion.Feed.SourceId = UMX.tag "messageDb" | ||
|
|
||
| module internal Npgsql = | ||
| let connect connectionString ct = task { | ||
| let conn = new NpgsqlConnection(connectionString) | ||
| do! conn.OpenAsync(ct) | ||
| return conn } |
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,121 @@ | ||
| namespace Propulsion.MessageDb | ||
|
|
||
| open FSharp.Control | ||
| open FsCodec | ||
| open FsCodec.Core | ||
| open NpgsqlTypes | ||
| open Propulsion.Feed | ||
| open Propulsion.Feed.Core | ||
| open Propulsion.Internal | ||
| open System | ||
| open System.Data.Common | ||
| open System.Diagnostics | ||
|
|
||
|
|
||
| module Core = | ||
| type MessageDbCategoryClient(connectionString) = | ||
| let connect = Npgsql.connect connectionString | ||
| let parseRow (reader: DbDataReader) = | ||
| let readNullableString idx = if reader.IsDBNull(idx) then None else Some (reader.GetString idx) | ||
| let streamName = reader.GetString(8) | ||
| let event = TimelineEvent.Create( | ||
| index = reader.GetInt64(0), | ||
| eventType = reader.GetString(1), | ||
| data = ReadOnlyMemory(Text.Encoding.UTF8.GetBytes(reader.GetString 2)), | ||
| meta = ReadOnlyMemory(Text.Encoding.UTF8.GetBytes(reader.GetString 3)), | ||
| eventId = reader.GetGuid(4), | ||
| ?correlationId = readNullableString 5, | ||
| ?causationId = readNullableString 6, | ||
| context = reader.GetInt64(9), | ||
| timestamp = DateTimeOffset(DateTime.SpecifyKind(reader.GetDateTime(7), DateTimeKind.Utc))) | ||
|
|
||
| struct(StreamName.parse streamName, event) | ||
| member _.ReadCategoryMessages(category: TrancheId, fromPositionInclusive: int64, batchSize: int, ct) = task { | ||
| use! conn = connect ct | ||
| let command = conn.CreateCommand(CommandText = "select position, type, data, metadata, id::uuid, | ||
| (metadata::jsonb->>'$correlationId')::text, | ||
| (metadata::jsonb->>'$causationId')::text, | ||
| time, stream_name, global_position | ||
| from get_category_messages(@Category, @Position, @BatchSize);") | ||
| command.Parameters.AddWithValue("Category", NpgsqlDbType.Text, TrancheId.toString category) |> ignore | ||
| command.Parameters.AddWithValue("Position", NpgsqlDbType.Bigint, fromPositionInclusive) |> ignore | ||
| command.Parameters.AddWithValue("BatchSize", NpgsqlDbType.Bigint, int64 batchSize) |> ignore | ||
|
|
||
| let mutable checkpoint = fromPositionInclusive | ||
|
|
||
| use! reader = command.ExecuteReaderAsync(ct) | ||
| let events = [| while reader.Read() do yield parseRow reader |] | ||
|
|
||
| checkpoint <- match Array.tryLast events with Some (_, ev) -> unbox<int64> ev.Context | None -> checkpoint | ||
|
|
||
| return { checkpoint = Position.parse checkpoint; items = events; isTail = events.Length = 0 } } | ||
| member _.ReadCategoryLastVersion(category: TrancheId, ct) = task { | ||
| use! conn = connect ct | ||
| let command = conn.CreateCommand(CommandText = "select max(global_position) from messages where category(stream_name) = @Category;") | ||
| command.Parameters.AddWithValue("Category", NpgsqlDbType.Text, TrancheId.toString category) |> ignore | ||
|
|
||
| use! reader = command.ExecuteReaderAsync(ct) | ||
| return if reader.Read() then reader.GetInt64(0) else 0L } | ||
|
|
||
| module private Impl = | ||
| open Core | ||
| open Propulsion.Infrastructure // AwaitTaskCorrect | ||
|
|
||
| let readBatch batchSize (store : MessageDbCategoryClient) (category, pos) : Async<Propulsion.Feed.Core.Batch<_>> = async { | ||
| let! ct = Async.CancellationToken | ||
| let positionInclusive = Position.toInt64 pos | ||
| let! x = store.ReadCategoryMessages(category, positionInclusive, batchSize, ct) |> Async.AwaitTaskCorrect | ||
| return x } | ||
|
|
||
| let readTailPositionForTranche (store : MessageDbCategoryClient) trancheId : Async<Propulsion.Feed.Position> = async { | ||
| let! ct = Async.CancellationToken | ||
| let! lastEventPos = store.ReadCategoryLastVersion(trancheId, ct) |> Async.AwaitTaskCorrect | ||
| return Position.parse lastEventPos } | ||
|
|
||
| type MessageDbSource | ||
| ( log : Serilog.ILogger, statsInterval, | ||
| client: Core.MessageDbCategoryClient, batchSize, tailSleepInterval, | ||
| checkpoints : Propulsion.Feed.IFeedCheckpointStore, sink : Propulsion.Streams.Default.Sink, | ||
| categories, | ||
| // Override default start position to be at the tail of the index. Default: Replay all events. | ||
| ?startFromTail, | ||
| ?sourceId) = | ||
| inherit Propulsion.Feed.Core.TailingFeedSource | ||
| ( log, statsInterval, defaultArg sourceId FeedSourceId.wellKnownId, tailSleepInterval, checkpoints, | ||
| ( if startFromTail <> Some true then None | ||
| else Some (Impl.readTailPositionForTranche client)), | ||
| sink, | ||
| (fun req -> asyncSeq { | ||
| let sw = Stopwatch.StartNew() | ||
| let! b = Impl.readBatch batchSize client req | ||
| yield sw.Elapsed, b }), | ||
| string) | ||
| new (log, statsInterval, connectionString, batchSize, tailSleepInterval, checkpoints, sink, trancheIds, ?startFromTail, ?sourceId) = | ||
| MessageDbSource(log, statsInterval, Core.MessageDbCategoryClient(connectionString), | ||
| batchSize, tailSleepInterval, checkpoints, sink, trancheIds, ?startFromTail=startFromTail, ?sourceId=sourceId) | ||
|
|
||
| abstract member ListTranches : unit -> Async<Propulsion.Feed.TrancheId array> | ||
| default _.ListTranches() = async { return categories |> Array.map TrancheId.parse } | ||
|
|
||
| abstract member Pump : unit -> Async<unit> | ||
| default x.Pump() = base.Pump(x.ListTranches) | ||
|
|
||
| abstract member Start : unit -> Propulsion.SourcePipeline<Propulsion.Feed.Core.FeedMonitor> | ||
| default x.Start() = base.Start(x.Pump()) | ||
|
|
||
|
|
||
| /// Pumps to the Sink until either the specified timeout has been reached, or all items in the Source have been fully consumed | ||
| member x.RunUntilCaughtUp(timeout : TimeSpan, statsInterval : IntervalTimer) = task { | ||
| let sw = Stopwatch.start () | ||
| use pipeline = x.Start() | ||
|
|
||
| try System.Threading.Tasks.Task.Delay(timeout).ContinueWith(fun _ -> pipeline.Stop()) |> ignore | ||
|
|
||
| let initialReaderTimeout = TimeSpan.FromMinutes 1. | ||
| do! pipeline.Monitor.AwaitCompletion(initialReaderTimeout, awaitFullyCaughtUp = true, logInterval = TimeSpan.FromSeconds 30) | ||
| pipeline.Stop() | ||
|
|
||
| if sw.ElapsedSeconds > 2 then statsInterval.Trigger() | ||
| // force a final attempt to flush anything not already checkpointed (normally checkpointing is at 5s intervals) | ||
| return! x.Checkpoint() | ||
| finally statsInterval.SleepUntilTriggerCleared() } |
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,28 @@ | ||
| <Project Sdk="Microsoft.NET.Sdk"> | ||
|
|
||
| <PropertyGroup> | ||
| <TargetFramework>net6.0</TargetFramework> | ||
| <GenerateDocumentationFile>true</GenerateDocumentationFile> | ||
| </PropertyGroup> | ||
|
|
||
| <ItemGroup> | ||
| <Compile Include="..\Propulsion\Infrastructure.fs"> | ||
| <Link>Infrastructure.fs</Link> | ||
| </Compile> | ||
| <Compile Include="Internal.fs" /> | ||
| <Compile Include="MessageDbSource.fs" /> | ||
| <Compile Include="ReaderCheckpoint.fs" /> | ||
| <Content Include="Readme.md" /> | ||
| </ItemGroup> | ||
|
|
||
| <ItemGroup> | ||
| <PackageReference Include="MinVer" Version="4.2.0" PrivateAssets="All" /> | ||
| <PackageReference Include="Npgsql" Version="6.0.7" /> | ||
| </ItemGroup> | ||
|
|
||
|
|
||
| <ItemGroup> | ||
| <ProjectReference Include="..\Propulsion.Feed\Propulsion.Feed.fsproj" /> | ||
| </ItemGroup> | ||
|
|
||
| </Project> |
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,73 @@ | ||
| module Propulsion.MessageDb.ReaderCheckpoint | ||
|
|
||
| open Npgsql | ||
| open NpgsqlTypes | ||
| open Propulsion.Feed | ||
| open Propulsion.Infrastructure | ||
|
|
||
|
|
||
| let table = "propulsion_checkpoint" | ||
|
|
||
| let createIfNotExists (conn : NpgsqlConnection, schema: string) = | ||
| let cmd = conn.CreateCommand(CommandText = $"create table if not exists {schema}.{table} ( | ||
| source text not null, | ||
| tranche text not null, | ||
| consumer_group text not null, | ||
| position bigint not null, | ||
| primary key (source, tranche, consumer_group));") | ||
| cmd.ExecuteNonQueryAsync() |> Async.AwaitTaskCorrect |> Async.Ignore<int> | ||
|
|
||
| let commitPosition (conn : NpgsqlConnection, schema: string) source tranche (consumerGroup : string) (position : int64) | ||
| = async { | ||
| let cmd = conn.CreateCommand(CommandText = $"insert into {schema}.{table}(source, tranche, consumer_group, position) | ||
| values (@Source, @Tranche, @ConsumerGroup, @Position) | ||
| on conflict (source, tranche, consumer_group) | ||
| do update set position = @Position;") | ||
| cmd.Parameters.AddWithValue("Source", NpgsqlDbType.Text, SourceId.toString source) |> ignore | ||
| cmd.Parameters.AddWithValue("Tranche", NpgsqlDbType.Text, TrancheId.toString tranche) |> ignore | ||
| cmd.Parameters.AddWithValue("ConsumerGroup", NpgsqlDbType.Text, consumerGroup) |> ignore | ||
| cmd.Parameters.AddWithValue("Position", NpgsqlDbType.Bigint, position) |> ignore | ||
|
|
||
| let! ct = Async.CancellationToken | ||
| do! cmd.ExecuteNonQueryAsync(ct) |> Async.AwaitTaskCorrect |> Async.Ignore<int> } | ||
|
|
||
| let tryGetPosition (conn : NpgsqlConnection, schema : string) source tranche (consumerGroup : string) = async { | ||
| let cmd = conn.CreateCommand(CommandText = $"select position from {schema}.{table} | ||
| where source = @Source | ||
| and tranche = @Tranche | ||
| and consumer_group = @ConsumerGroup") | ||
| cmd.Parameters.AddWithValue("Source", NpgsqlDbType.Text, SourceId.toString source) |> ignore | ||
| cmd.Parameters.AddWithValue("Tranche", NpgsqlDbType.Text, TrancheId.toString tranche) |> ignore | ||
| cmd.Parameters.AddWithValue("ConsumerGroup", NpgsqlDbType.Text, consumerGroup) |> ignore | ||
|
|
||
| let! ct = Async.CancellationToken | ||
| use! reader = cmd.ExecuteReaderAsync(ct) |> Async.AwaitTaskCorrect | ||
| return if reader.Read() then ValueSome (reader.GetInt64 0) else ValueNone } | ||
|
|
||
| type CheckpointStore(connString : string, schema: string, consumerGroupName, defaultCheckpointFrequency) = | ||
| let connect = Npgsql.connect connString | ||
|
|
||
| member _.CreateSchemaIfNotExists() = async { | ||
| let! ct = Async.CancellationToken | ||
| use! conn = connect ct |> Async.AwaitTaskCorrect | ||
| return! createIfNotExists (conn, schema) } | ||
|
|
||
| interface IFeedCheckpointStore with | ||
|
|
||
| member _.Start(source, tranche, ?establishOrigin) = async { | ||
| let! ct = Async.CancellationToken | ||
| use! conn = connect ct |> Async.AwaitTaskCorrect | ||
| let! maybePos = tryGetPosition (conn, schema) source tranche consumerGroupName | ||
| let! pos = | ||
| match maybePos, establishOrigin with | ||
| | ValueSome pos, _ -> async { return Position.parse pos } | ||
| | ValueNone, Some f -> f | ||
| | ValueNone, None -> async { return Position.initial } | ||
| return defaultCheckpointFrequency, pos } | ||
|
|
||
| member _.Commit(source, tranche, pos) = async { | ||
| let! ct = Async.CancellationToken | ||
| use! conn = connect ct |> Async.AwaitTaskCorrect | ||
| return! commitPosition (conn, schema) source tranche consumerGroupName (Position.toInt64 pos) } | ||
|
|
||
|
|
Oops, something went wrong.