Aggregate layout/naming consistency

Author: bartelink

equinox-testbed/Program.fs

@@ -69,11 +69,11 @@ module CmdParser =
         member __.Options =             a.GetResults Cached @ a.GetResults Unfolds
         member __.Cache =               __.Options |> List.contains Cached
         member __.Unfolds =             __.Options |> List.contains Unfolds
-        member __.BatchSize =           a.GetResult(BatchSize,500)
-        member __.Test =                a.GetResult(Name,Tests.Favorite)
-        member __.ErrorCutoff =         a.GetResult(ErrorCutoff,10000L)
-        member __.TestsPerSecond =      a.GetResult(TestsPerSecond,100)
-        member __.Duration =            a.GetResult(DurationM,30.) |> TimeSpan.FromMinutes
+        member __.BatchSize =           a.GetResult(BatchSize, 500)
+        member __.Test =                a.GetResult(Name, Tests.Favorite)
+        member __.ErrorCutoff =         a.GetResult(ErrorCutoff, 10000L)
+        member __.TestsPerSecond =      a.GetResult(TestsPerSecond, 100)
+        member __.Duration =            a.GetResult(DurationM, 30.) |> TimeSpan.FromMinutes
         member __.ReportingIntervals =
             match a.GetResults(ReportIntervalS) with
             | [] -> TimeSpan.FromSeconds 10.|> Seq.singleton
@@ -90,13 +90,13 @@ module CmdParser =
             | Some (Es sargs) ->
                 let storeLog = createStoreLog <| sargs.Contains Storage.EventStore.Parameters.VerboseStore
                 log.Information("Running transactions in-process against EventStore with storage options: {options:l}", __.Options)
-                storeLog, Storage.EventStore.config (log,storeLog) (__.Cache, __.Unfolds, __.BatchSize) sargs
+                storeLog, Storage.EventStore.config (log, storeLog) (__.Cache, __.Unfolds, __.BatchSize) sargs
 //#endif
 //#if cosmos
             | Some (Cosmos sargs) ->
                 let storeLog = createStoreLog <| sargs.Contains Storage.Cosmos.Parameters.VerboseStore
                 log.Information("Running transactions in-process against CosmosDb with storage options: {options:l}", __.Options)
-                storeLog, Storage.Cosmos.config (log,storeLog) (__.Cache, __.Unfolds, __.BatchSize) (Storage.Cosmos.Arguments sargs)
+                storeLog, Storage.Cosmos.config (log, storeLog) (__.Cache, __.Unfolds, __.BatchSize) (Storage.Cosmos.Arguments sargs)
 //#endif
 #if ((!cosmos && !eventStore) || (cosmos && eventStore))
             | _ -> raise <| Storage.MissingArg (sprintf "Please identify a valid store: memory, es, cosmos")
@@ -140,7 +140,7 @@ module LoadTest =
                 with e -> domainLog.Warning(e, "Test threw an exception"); e.Reraise () }
         execute
     let private createResultLog fileName = LoggerConfiguration().WriteTo.File(fileName).CreateLogger()
-    let run (log: ILogger) (verbose,verboseConsole,maybeSeq) reportFilename (a : CmdParser.TestArguments) =
+    let run (log: ILogger) (verbose, verboseConsole, maybeSeq) reportFilename (a : CmdParser.TestArguments) =
         let createStoreLog verboseStore = createStoreLog verboseStore verboseConsole maybeSeq
         let storeLog, storeConfig: ILogger * Storage.StorageConfig = a.ConfigureStore(log, createStoreLog)
         let runSingleTest : ClientId -> Async<unit> =
@@ -200,8 +200,8 @@ let main argv =
             let maybeSeq = if args.Contains LocalSeq then Some "http://localhost:5341" else None
             let verbose = args.Contains Verbose
             let log = createDomainLog verbose verboseConsole maybeSeq
-            let reportFilename = args.GetResult(LogFile,programName+".log") |> fun n -> System.IO.FileInfo(n).FullName
-            LoadTest.run log (verbose,verboseConsole,maybeSeq) reportFilename (TestArguments rargs)
+            let reportFilename = args.GetResult(LogFile, programName+".log") |> fun n -> System.IO.FileInfo(n).FullName
+            LoadTest.run log (verbose, verboseConsole, maybeSeq) reportFilename (TestArguments rargs)
         | _ -> failwith "Please specify a valid subcommand :- run"
         0 
     with :? Argu.ArguParseException as e -> eprintfn "%s" e.Message; 1

equinox-testbed/Services.fs

@@ -19,8 +19,9 @@ module Domain =
                 | Unfavorited                           of Unfavorited
                 interface TypeShape.UnionContract.IUnionContract
             let codec = FsCodec.NewtonsoftJson.Codec.Create<Event>()
+            let (|For|) (id : ClientId) = Equinox.AggregateId("Favorites", ClientId.toString id)
 
-        module Folds =
+        module Fold =
 
             type State = Events.Favorited []
 
@@ -50,44 +51,42 @@ module Domain =
             | Favorite      of date : System.DateTimeOffset * skuIds : SkuId list
             | Unfavorite    of skuId : SkuId
 
-        module Commands =
-            let interpret command (state : Folds.State) =
-                let doesntHave skuId = state |> Array.exists (fun x -> x.skuId = skuId) |> not
-                match command with
-                | Favorite (date = date; skuIds = skuIds) ->
-                    [ for skuId in Seq.distinct skuIds do
-                        if doesntHave skuId then
-                            yield Events.Favorited { date = date; skuId = skuId } ]
-                | Unfavorite skuId ->
-                    if doesntHave skuId then [] else
-                    [ Events.Unfavorited { skuId = skuId } ]
-
-        type Service(log, resolveStream, ?maxAttempts) =
-
-            let (|AggregateId|) (id: ClientId) = Equinox.AggregateId("Favorites", ClientId.toString id)
-            let (|Stream|) (AggregateId id) = Equinox.Stream(log, resolveStream id, defaultArg maxAttempts 2)
-            let execute (Stream stream) command : Async<unit> =
-                stream.Transact(Commands.interpret command)
-            let read (Stream stream) : Async<Events.Favorited []> =
-                stream.Query id
+        let interpret command (state : Fold.State) =
+            let doesntHave skuId = state |> Array.exists (fun x -> x.skuId = skuId) |> not
+            match command with
+            | Favorite (date = date; skuIds = skuIds) ->
+                [ for skuId in Seq.distinct skuIds do
+                    if doesntHave skuId then
+                        yield Events.Favorited { date = date; skuId = skuId } ]
+            | Unfavorite skuId ->
+                if doesntHave skuId then [] else
+                [ Events.Unfavorited { skuId = skuId } ]
+
+        type Service internal (log, resolve, maxAttempts) =
+
+            let resolve (Events.For id) = Equinox.Stream<Events.Event, Fold.State>(log, resolve id, maxAttempts)
 
             member __.Execute(clientId, command) =
-                execute clientId command
+                let stream = resolve clientId
+                stream.Transact(interpret command)
 
-            member __.Favorite(clientId, skus) =
-                execute clientId (Command.Favorite(DateTimeOffset.Now, skus))
+            member x.Favorite(clientId, skus) =
+                x.Execute(clientId, Command.Favorite (DateTimeOffset.Now, skus))
 
-            member __.Unfavorite(clientId, skus) =
-                execute clientId (Command.Unfavorite skus)
+            member x.Unfavorite(clientId, skus) =
+                x.Execute(clientId, Command.Unfavorite skus)
 
             member __.List clientId : Async<Events.Favorited []> =
-                read clientId
-            
+                let stream = resolve clientId
+                stream.Query id
+
+        let create log resolve = Service(log, resolve, maxAttempts = 3)
+
 open Microsoft.Extensions.DependencyInjection
 
 type StreamResolver(storage) =
     member __.Resolve
-        (   codec : FsCodec.IUnionEncoder<'event,byte[],_>,
+        (   codec : FsCodec.IUnionEncoder<'event, byte[], _>,
             fold: ('state -> 'event seq -> 'state),
             initial: 'state,
             snapshot: (('event -> bool) * ('state -> 'event))) =
@@ -99,22 +98,22 @@ type StreamResolver(storage) =
 //#if eventStore
         | Storage.StorageConfig.Es (gateway, caching, unfolds) ->
             let accessStrategy = if unfolds then Equinox.EventStore.AccessStrategy.RollingSnapshots snapshot |> Some else None
-            Equinox.EventStore.Resolver<'event,'state,_>(gateway, codec, fold, initial, ?caching = caching, ?access = accessStrategy).Resolve
+            Equinox.EventStore.Resolver<'event, 'state, _>(gateway, codec, fold, initial, ?caching = caching, ?access = accessStrategy).Resolve
 //#endif
 //#if cosmos
         | Storage.StorageConfig.Cosmos (gateway, caching, unfolds, databaseId, containerId) ->
             let store = Equinox.Cosmos.Context(gateway, databaseId, containerId)
             let accessStrategy = if unfolds then Equinox.Cosmos.AccessStrategy.Snapshot snapshot else Equinox.Cosmos.AccessStrategy.Unoptimized
-            Equinox.Cosmos.Resolver<'event,'state,_>(store, codec, fold, initial, caching, accessStrategy).Resolve
+            Equinox.Cosmos.Resolver<'event, 'state, _>(store, codec, fold, initial, caching, accessStrategy).Resolve
 //#endif
 
 type ServiceBuilder(storageConfig, handlerLog) =
      let resolver = StreamResolver(storageConfig)
 
      member __.CreateFavoritesService() =
-        let fold, initial = Domain.Favorites.Folds.fold, Domain.Favorites.Folds.initial
-        let snapshot = Domain.Favorites.Folds.isOrigin,Domain.Favorites.Folds.compact
-        Domain.Favorites.Service(handlerLog, resolver.Resolve(Domain.Favorites.Events.codec,fold,initial,snapshot))
+        let fold, initial = Domain.Favorites.Fold.fold, Domain.Favorites.Fold.initial
+        let snapshot = Domain.Favorites.Fold.isOrigin, Domain.Favorites.Fold.compact
+        Domain.Favorites.create handlerLog (resolver.Resolve(Domain.Favorites.Events.codec, fold, initial, snapshot))
 
 let register (services : IServiceCollection, storageConfig, handlerLog) =
     let regF (factory : IServiceProvider -> 'T) = services.AddSingleton<'T>(fun (sp: IServiceProvider) -> factory sp) |> ignore

equinox-testbed/Storage.fs

@@ -59,13 +59,13 @@ module Cosmos =
                 | Database _ ->         "specify a database name for store. (optional if environment variable EQUINOX_COSMOS_DATABASE specified)"
                 | Container _ ->        "specify a container name for store. (optional if environment variable EQUINOX_COSMOS_CONTAINER specified)"
     type Arguments(a : ParseResults<Parameters>) =
-        member __.Mode =                a.GetResult(ConnectionMode,Equinox.Cosmos.ConnectionMode.Direct)
+        member __.Mode =                a.GetResult(ConnectionMode, Equinox.Cosmos.ConnectionMode.Direct)
         member __.Connection =          a.TryGetResult Connection |> defaultWithEnvVar "EQUINOX_COSMOS_CONNECTION" "Connection"
         member __.Database =            a.TryGetResult Database   |> defaultWithEnvVar "EQUINOX_COSMOS_DATABASE"   "Database"
         member __.Container =           a.TryGetResult Container  |> defaultWithEnvVar "EQUINOX_COSMOS_CONTAINER"  "Container"
 
-        member __.Timeout =             a.GetResult(Timeout,5.) |> TimeSpan.FromSeconds
-        member __.Retries =             a.GetResult(Retries,1)
+        member __.Timeout =             a.GetResult(Timeout, 5.) |> TimeSpan.FromSeconds
+        member __.Retries =             a.GetResult(Retries, 1)
         member __.MaxRetryWaitTime =    a.GetResult(RetriesWaitTime, 5.) |> TimeSpan.FromSeconds
 
     /// Standing up an Equinox instance is necessary to run for test purposes; You'll need to either:
@@ -77,7 +77,7 @@ module Cosmos =
 
     let private createGateway connection maxItems = Gateway(connection, BatchingPolicy(defaultMaxItems=maxItems))
     let private context (log: ILogger, storeLog: ILogger) (a : Arguments) =
-        let (Discovery.UriAndKey (endpointUri,_)) as discovery = a.Connection|> Discovery.FromConnectionString
+        let (Discovery.UriAndKey (endpointUri, _)) as discovery = a.Connection|> Discovery.FromConnectionString
         log.Information("CosmosDb {mode} {connection} Database {database} Container {container}",
             a.Mode, endpointUri, a.Database, a.Container)
         Log.Information("CosmosDb timeout {timeout}s; Throttling retries {retries}, max wait {maxRetryWaitTime}s",
@@ -123,13 +123,13 @@ module EventStore =
     open Equinox.EventStore
 
     type Arguments(a : ParseResults<Parameters>) =
-        member __.Host =                a.GetResult(Host,"localhost")
-        member __.Credentials =         a.GetResult(Username,"admin"), a.GetResult(Password,"changeit")
+        member __.Host =                a.GetResult(Host, "localhost")
+        member __.Credentials =         a.GetResult(Username, "admin"), a.GetResult(Password, "changeit")
 
         member __.Retries =             a.GetResult(Retries, 1)
-        member __.Timeout =             a.GetResult(Timeout,5.) |> TimeSpan.FromSeconds
-        member __.HeartbeatTimeout =    a.GetResult(HeartbeatTimeout,1.5) |> float |> TimeSpan.FromSeconds
-        member __.ConcurrentOperationsLimit = a.GetResult(ConcurrentOperationsLimit,5000)
+        member __.Timeout =             a.GetResult(Timeout, 5.) |> TimeSpan.FromSeconds
+        member __.HeartbeatTimeout =    a.GetResult(HeartbeatTimeout, 1.5) |> float |> TimeSpan.FromSeconds
+        member __.ConcurrentOperationsLimit = a.GetResult(ConcurrentOperationsLimit, 5000)
 
     let private connect (log: Serilog.ILogger) (dnsQuery, heartbeatTimeout, col) (username, password) (operationTimeout, operationRetries) =
         Connector(username, password, reqTimeout=operationTimeout, reqRetries=operationRetries,

equinox-testbed/Tests.fs

@@ -14,6 +14,6 @@ let executeLocal (container: ServiceProvider) test: ClientId -> Async<unit> =
         let service = container.GetRequiredService<Services.Domain.Favorites.Service>()
         fun clientId -> async {
             let sku = mkSkuId ()
-            do! service.Favorite(clientId,[sku])
+            do! service.Favorite(clientId, [sku])
             let! items = service.List clientId
             if items |> Array.exists (fun x -> x.skuId = sku) |> not then invalidOp "Added item not found" }

equinox-web-csharp/Domain/Aggregate.cs

@@ -36,6 +36,7 @@ public static Event TryDecode(string et, byte[] json)
             }
 
             public static Tuple<string, byte[]> Encode(Event e) => Tuple.Create(e.GetType().Name, Codec.Encode(e));
+            public static Target For(ClientId id) => Target.NewAggregateId("Aggregate", id.ToString());
         }
         public class State
         {
@@ -94,18 +95,18 @@ public static IEnumerable<Event> Interpret(State s, Command x)
 
         class Handler
         {
-            readonly EquinoxStream<Event, State> _inner;
+            readonly EquinoxStream<Event, State> _stream;
 
             public Handler(ILogger log, IStream<Event, State> stream) =>
-                _inner = new EquinoxStream<Event, State>(State.Fold, log, stream);
+                _stream = new EquinoxStream<Event, State>(State.Fold, log, stream);
 
             /// Execute `command`, syncing any events decided upon
             public Task<Unit> Execute(Command c) =>
-                _inner.Execute(s => Command.Interpret(s, c));
+                _stream.Execute(s => Command.Interpret(s, c));
 
             /// Establish the present state of the Stream, project from that as specified by `projection`
             public Task<T> Query<T>(Func<State, T> projection) =>
-                _inner.Query(projection);
+                _stream.Query(projection);
         }
 
         public class View
@@ -116,20 +117,18 @@ public class View
         public class Service
         {
             /// Maps a ClientId to Handler for the relevant stream
-            readonly Func<string, Handler> _stream;
-
-            static Target AggregateId(string id) => Target.NewAggregateId("Aggregate", id);
+            readonly Func<ClientId, Handler> _stream;
 
             public Service(ILogger handlerLog, Func<Target, IStream<Event, State>> resolve) =>
-                _stream = id => new Handler(handlerLog, resolve(AggregateId(id)));
+                _stream = id => new Handler(handlerLog, resolve(Event.For(id)));
 
             /// Execute the specified command 
-            public Task<Unit> Execute(string id, Command command) =>
+            public Task<Unit> Execute(ClientId id, Command command) =>
                 _stream(id).Execute(command);
 
             /// Read the present state
             // TOCONSIDER: you should probably be separating this out per CQRS and reading from a denormalized/cached set of projections
-            public Task<View> Read(string id) => _stream(id).Query(Render);
+            public Task<View> Read(ClientId id) => _stream(id).Query(Render);
 
             static View Render(State s) => new View() {Sorted = s.Happened};
         }

equinox-web-csharp/Domain/Infrastructure.cs

@@ -31,15 +31,15 @@ public EquinoxStream(
         }
 
         /// Run the decision method, letting it decide whether or not the Command's intent should manifest as Events
-        public async Task<Unit> Execute(Func<TState,IEnumerable<TEvent>> interpret)
+        public async Task<Unit> Execute(Func<TState, IEnumerable<TEvent>> interpret)
         {
             FSharpList<TEvent> decide_(TState state)
             {
                 var a = new Accumulator<TEvent, TState>(FuncConvert.FromFunc(_fold), state);
                 a.Execute(interpret);
                 return a.Accumulated;
             }
-            return await FSharpAsync.StartAsTask(Transact(FuncConvert.FromFunc<TState,FSharpList<TEvent>>(decide_)), null, null);
+            return await FSharpAsync.StartAsTask(Transact(FuncConvert.FromFunc<TState, FSharpList<TEvent>>(decide_)), null, null);
         }
 
         /// Execute a command, as Decide(Action) does, but also yield an outcome from the decision
@@ -51,7 +51,7 @@ Tuple<T, FSharpList<TEvent>> decide_(TState state)
                 var r = decide(a);
                 return Tuple.Create(r, a.Accumulated);
             }
-            return await FSharpAsync.StartAsTask<T>(Transact(FuncConvert.FromFunc<TState,Tuple<T,FSharpList<TEvent>>>(decide_)), null, null);
+            return await FSharpAsync.StartAsTask<T>(Transact(FuncConvert.FromFunc<TState, Tuple<T, FSharpList<TEvent>>>(decide_)), null, null);
         }
 
         // Project from the synchronized state, without the possibility of adding events that Decide(Func) admits