Target Propulsion 2.13.0-beta.4

oskardudycz · May 23, 2022 · 9e9d873 · 9e9d873
1 parent 448926e
commit 9e9d873
Show file tree

Hide file tree

Showing 11 changed files with 38 additions and 27 deletions.
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/ConfirmedEpoch.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/ConfirmedEpoch.fs
@@ -92,6 +92,9 @@ module Config =
         | Config.Store.Cosmos (context, cache) ->
             let cat = Config.Cosmos.createUnoptimized Events.codecJsonElement Fold.initial Fold.fold (context, cache)
             cat.Resolve
+        | Config.Store.Dynamo (context, cache) ->
+            let cat = Config.Dynamo.createUnoptimized Events.codec Fold.initial Fold.fold (context, cache)
+            cat.Resolve
     let private resolveDecider store = streamName >> resolveStream store >> Config.createDecider
     let shouldClose maxItemsPerEpoch candidateItems currentItems = Array.length currentItems + Array.length candidateItems >= maxItemsPerEpoch
     let create maxItemsPerEpoch = resolveDecider >> create_ (shouldClose maxItemsPerEpoch)

diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/ECommerce.Domain.fsproj b/Sample/ECommerce.Equinox/ECommerce.Domain/ECommerce.Domain.fsproj
@@ -18,12 +18,12 @@
     </ItemGroup>
 
     <ItemGroup>
-        <PackageReference Include="Equinox.CosmosStore" Version="4.0.0-beta.4.2" />
-        <PackageReference Include="Equinox.DynamoStore" Version="4.0.0-beta.4.2" />
-        <PackageReference Include="Equinox.EventStoreDb" Version="4.0.0-beta.4.2" />
-        <PackageReference Include="Equinox.MemoryStore" Version="4.0.0-beta.4.2" />
+        <PackageReference Include="Equinox.CosmosStore" Version="4.0.0-beta.4.5" />
+        <PackageReference Include="Equinox.DynamoStore" Version="4.0.0-beta.4.5" />
+        <PackageReference Include="Equinox.EventStoreDb" Version="4.0.0-beta.4.5" />
+        <PackageReference Include="Equinox.MemoryStore" Version="4.0.0-beta.4.5" />
         <PackageReference Include="FsCodec.SystemTextJson" Version="3.0.0-rc.2.2" />
-        <PackageReference Include="Propulsion" Version="2.13.0-beta.3" />
+        <PackageReference Include="Propulsion" Version="2.13.0-beta.4" />
     </ItemGroup>
 
 </Project>
diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/ShoppingCart.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/ShoppingCart.fs
@@ -166,6 +166,9 @@ module Config =
         | Config.Store.Cosmos (context, cache) ->
             let cat = Config.Cosmos.createUnoptimized Events.codecJsonElement Fold.initial Fold.fold (context, cache)
             cat.Resolve
+        | Config.Store.Dynamo (context, cache) ->
+            let cat = Config.Dynamo.createUnoptimized Events.codec Fold.initial Fold.fold (context, cache)
+            cat.Resolve
     let private resolveDecider store = streamName >> resolveStream store >> Config.createDecider
     let create_ pricer store =
         Service(resolveDecider store, calculatePrice pricer)

diff --git a/Sample/ECommerce.Equinox/ECommerce.Domain/ShoppingCartSummary.fs b/Sample/ECommerce.Equinox/ECommerce.Domain/ShoppingCartSummary.fs
@@ -72,5 +72,8 @@ module Config =
         | Config.Store.Cosmos (context, cache) ->
             let cat = Config.Cosmos.createRollingState Events.codecJsonElement Fold.initial Fold.fold Fold.toSnapshot (context, cache)
             cat.Resolve
+        | Config.Store.Dynamo (context, cache) ->
+            let cat = Config.Dynamo.createRollingState Events.codec Fold.initial Fold.fold Fold.toSnapshot (context, cache)
+            cat.Resolve
     let private resolveDecider store = streamName >> resolveStream store >> Config.createDecider
     let create = resolveDecider >> Service
diff --git a/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Program.fs b/Sample/ECommerce.Equinox/ECommerce.FeedConsumer/Program.fs
@@ -79,7 +79,7 @@ module Args =
             | Args.StoreArguments.Esdb a ->
                 let context = a.Connect(Log.Logger, AppName, EventStore.Client.NodePreference.Leader) |> EventStoreContext.create
                 Config.Store.Esdb (context, cache)
-        member x.CheckpointStoreConfig(mainStore) : CheckpointStore.Config =
+        member private x.CheckpointStoreConfig(mainStore) : CheckpointStore.Config =
             match mainStore with
             | Config.Store.Cosmos (context, cache) -> CheckpointStore.Config.Cosmos (context, cache)
             | Config.Store.Dynamo (context, cache) -> CheckpointStore.Config.Dynamo (context, cache)
@@ -95,8 +95,9 @@ module Args =
                         let context = a.Connect() |> DynamoStoreContext.create
                         CheckpointStore.Config.Dynamo (context, cache)
                  | _ -> failwith "unexpected"
-        member x.CreateCheckpointStore(config) : Propulsion.Feed.IFeedCheckpointStore =
-            CheckpointStore.create (x.ConsumerGroupName, x.CheckpointInterval) Config.log config
+        member x.CreateCheckpointStore(mainStore) : Propulsion.Feed.IFeedCheckpointStore =
+            let config = x.CheckpointStoreConfig(mainStore)
+            CheckpointStore.create (x.ConsumerGroupName, x.CheckpointInterval) ECommerce.Domain.Config.log config
 
     /// Parse the commandline; can throw exceptions in response to missing arguments and/or `-h`/`--help` args
     let parse tryGetConfigValue argv =
@@ -113,7 +114,7 @@ let build (args : Args.Arguments) =
         let stats = Ingester.Stats(log, args.StatsInterval, args.StateInterval, logExternalStats = args.DumpStoreMetrics)
         Propulsion.Streams.StreamsProjector.Start(log, args.MaxReadAhead, args.FcsDop, handle, stats, args.StatsInterval)
     let pumpSource =
-        let checkpoints = args.CreateCheckpointStore(args.CheckpointStoreConfig store)
+        let checkpoints = args.CreateCheckpointStore(store)
         let feed = ApiClient.TicketsFeed args.BaseUri
         let source =
             Propulsion.Feed.FeedSource(

diff --git a/Sample/ECommerce.Equinox/ECommerce.Infrastructure/Args.fs b/Sample/ECommerce.Equinox/ECommerce.Infrastructure/Args.fs
@@ -96,7 +96,7 @@ module Dynamo =
         let table =                         a.TryGetResult Table      |> Option.defaultWith (fun () -> c.DynamoTable)
         let retries =                       a.GetResult(Retries, 1)
         let timeout =                       a.GetResult(RetriesTimeoutS, 5.) |> TimeSpan.FromSeconds
-        let connector =                     Equinox.DynamoStore.DynamoStoreConnector(serviceUrl, accessKey, secretKey, retries, timeout)
+        let connector =                     Equinox.DynamoStore.DynamoStoreConnector(serviceUrl, accessKey, secretKey, timeout, retries)
         member val Verbose =                a.Contains Verbose
         member _.Connect() =                connector.ConnectStore(connector.CreateClient(), "Main", table)
 

diff --git a/Sample/ECommerce.Equinox/ECommerce.Infrastructure/ECommerce.Infrastructure.fsproj b/Sample/ECommerce.Equinox/ECommerce.Infrastructure/ECommerce.Infrastructure.fsproj
@@ -15,12 +15,12 @@
     <ItemGroup>
         <PackageReference Include="Argu" Version="6.1.1" />
         <PackageReference Include="Destructurama.FSharp" Version="1.2.0" />
-        <PackageReference Include="Equinox.CosmosStore.Prometheus" Version="4.0.0-beta.4.2" />
-        <PackageReference Include="Equinox.DynamoStore.Prometheus" Version="4.0.0-beta.4.2" />
+        <PackageReference Include="Equinox.CosmosStore.Prometheus" Version="4.0.0-beta.4.5" />
+        <PackageReference Include="Equinox.DynamoStore.Prometheus" Version="4.0.0-beta.4.5" />
         <PackageReference Include="prometheus-net.AspNetCore" Version="3.6.0" />
-        <PackageReference Include="Propulsion.CosmosStore" Version="2.13.0-beta.2" />
-        <PackageReference Include="Propulsion.EventStoreDb" Version="2.13.0-beta.2" />
-        <PackageReference Include="Propulsion.DynamoStore" Version="2.13.0-beta.2" />
+        <PackageReference Include="Propulsion.CosmosStore" Version="2.13.0-beta.4" />
+        <PackageReference Include="Propulsion.EventStoreDb" Version="2.13.0-beta.4" />
+        <PackageReference Include="Propulsion.DynamoStore" Version="2.13.0-beta.4" />
         <PackageReference Include="Serilog.Sinks.Async" Version="1.5.0" />
         <PackageReference Include="Serilog.Sinks.Console" Version="4.0.0" />
     </ItemGroup>

diff --git a/Sample/ECommerce.Equinox/ECommerce.Infrastructure/Infrastructure.fs b/Sample/ECommerce.Equinox/ECommerce.Infrastructure/Infrastructure.fs
@@ -52,8 +52,8 @@ module CosmosStoreContext =
 
 type Equinox.DynamoStore.DynamoStoreClient with
 
-    member x.LogConfiguration(role : string, ?log) =
-        (defaultArg log Log.Logger).Information("DynamoStore {role:l} Table {table}", role) // TODO next ver has: , x.TableName)
+    member x.LogConfiguration(role, ?log) =
+        (defaultArg log Log.Logger).Information("DynamoStore {role:l} Table {table}", role, x.TableName)
 
 type Equinox.DynamoStore.DynamoStoreConnector with
 

diff --git a/Sample/ECommerce.Equinox/ECommerce.Infrastructure/SourceArgs.fs b/Sample/ECommerce.Equinox/ECommerce.Infrastructure/SourceArgs.fs
@@ -108,7 +108,7 @@ module Dynamo =
         let streamsDop =                    a.GetResult(StreamsDop, 4)
         let timeout =                       a.GetResult(RetriesTimeoutS, 60.) |> TimeSpan.FromSeconds
         let retries =                       a.GetResult(Retries, 9)
-        let connector =                     Equinox.DynamoStore.DynamoStoreConnector(serviceUrl, accessKey, secretKey, retries, timeout)
+        let connector =                     Equinox.DynamoStore.DynamoStoreConnector(serviceUrl, accessKey, secretKey, timeout, retries)
         let client =                        connector.CreateClient()
         member val Verbose =                a.Contains Verbose
         member _.Connect() =                let mainClient = connector.ConnectStore(client, "Main", table)

diff --git a/Sample/ECommerce.Equinox/ECommerce.Reactor/Program.fs b/Sample/ECommerce.Equinox/ECommerce.Reactor/Program.fs
@@ -51,7 +51,7 @@ module Args =
         member x.DumpStoreMetrics =         SourceArgs.dumpMetrics x.SourceArgs
         member val CheckpointInterval =     TimeSpan.FromHours 1.
         member val CacheSizeMb =            10
-        member x.CheckpointStoreConfig(mainStore : ECommerce.Domain.Config.Store<_>) : CheckpointStore.Config =
+        member private x.CheckpointStoreConfig(mainStore : ECommerce.Domain.Config.Store<_>) : CheckpointStore.Config =
             match mainStore with
             | ECommerce.Domain.Config.Store.Cosmos (context, cache) -> CheckpointStore.Config.Cosmos (context, cache)
             | ECommerce.Domain.Config.Store.Dynamo (context, cache) -> CheckpointStore.Config.Dynamo (context, cache)
@@ -67,7 +67,8 @@ module Args =
                         let context = a.Connect() |> DynamoStoreContext.create
                         CheckpointStore.Config.Dynamo (context, cache)
                  | _ -> failwith "unexpected"
-        member x.CreateCheckpointStore(config) : Propulsion.Feed.IFeedCheckpointStore =
+        member x.CreateCheckpointStore(mainStore) : Propulsion.Feed.IFeedCheckpointStore =
+            let config = x.CheckpointStoreConfig(mainStore)
             CheckpointStore.create (x.Group, x.CheckpointInterval) ECommerce.Domain.Config.log config
 
     /// Parse the commandline; can throw exceptions in response to missing arguments and/or `-h`/`--help` args
@@ -94,8 +95,8 @@ let build (args : Args.Arguments) =
         let handle = Reactor.Config.create (store, store)
         let sink = buildSink args log handle
         let source =
-            let parseFeedDoc : _ -> _ = Seq.collect Propulsion.CosmosStore.EquinoxSystemTextJsonParser.enumStreamEvents
-                                        >> Seq.filter (fun {stream = s } -> Reactor.isReactionStream s)
+            let parseFeedDoc = Seq.collect Propulsion.CosmosStore.EquinoxSystemTextJsonParser.enumStreamEvents
+                               >> Seq.filter (fun {stream = s } -> Reactor.isReactionStream s)
             let observer = Propulsion.CosmosStore.CosmosStoreSource.CreateObserver(log, sink.StartIngester, parseFeedDoc)
             let leases, startFromTail, maxItems, lagFrequency = a.MonitoringParams(log)
             Propulsion.CosmosStore.CosmosStoreSource.Start(log, monitored, leases, args.Group, observer, startFromTail, ?maxItems = maxItems, lagReportFreq = lagFrequency)
@@ -105,7 +106,7 @@ let build (args : Args.Arguments) =
         let context = storeClient |> DynamoStoreContext.create
         let sourceId = Propulsion.DynamoStore.FeedSourceId.wellKnownId
         let store = Config.Store.Dynamo (context, cache)
-        let checkpoints = args.CreateCheckpointStore(args.CheckpointStoreConfig store)
+        let checkpoints = args.CreateCheckpointStore(store)
         match a.MaybeOverrideRequested() with
         | None -> ()
         | Some (trancheId, group, pos) -> Async.RunSynchronously <| async {
@@ -115,7 +116,7 @@ let build (args : Args.Arguments) =
         let sink = buildSink args log handle
         let source =
             let indexClient, startFromTail,  maxItems, streamsDop = a.MonitoringParams(log)
-            let loadMode = Propulsion.DynamoStore.LoadMode.WithBodies (Reactor.isReactionStream, streamsDop, streamsContext)
+            let loadMode = Propulsion.DynamoStore.LoadMode.Hydrated (Reactor.isReactionStream, streamsDop, streamsContext)
             Propulsion.DynamoStore.DynamoStoreSource(
                 log, args.StatsInterval,
                 indexClient, sourceId, maxItems, TimeSpan.FromSeconds 0.5,
@@ -127,7 +128,7 @@ let build (args : Args.Arguments) =
         let context = conn |> EventStoreContext.create
         let sourceId = Propulsion.EventStoreDb.FeedSourceId.wellKnownId
         let store = Config.Store.Esdb (context, cache)
-        let checkpoints = args.CreateCheckpointStore(args.CheckpointStoreConfig store)
+        let checkpoints = args.CreateCheckpointStore(store)
         // TODO implement checkpoint reset mechanism
         let handle = Reactor.Config.create (store, store)
         let sink = buildSink args log handle
@@ -148,7 +149,7 @@ let run (args : Args.Arguments) = async {
             Async.AwaitKeyboardInterruptAsTaskCancelledException()
             source.AwaitWithStopOnCancellation()
             sink.AwaitWithStopOnCancellation() ]
-    return! Async.Parallel (build args) |> Async.Ignore<unit array> }
+    return! Async.Parallel work |> Async.Ignore<unit array> }
 
 [<EntryPoint>]
 let main argv =

diff --git a/Sample/ECommerce.Equinox/ECommerce.Tests/ECommerce.Tests.fsproj b/Sample/ECommerce.Equinox/ECommerce.Tests/ECommerce.Tests.fsproj
@@ -14,7 +14,7 @@
         <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.11.0"/>
 
         <PackageReference Include="FsCheck.Xunit" Version="3.0.0-beta1"/>
-        <PackageReference Include="unquote" Version="5.0.0"/>
+        <PackageReference Include="unquote" Version="6.1.0"/>
         <PackageReference Include="xunit" Version="2.4.1"/>
         <PackageReference Include="xunit.runner.visualstudio" Version="2.4.3">
             <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>