Helper naming consistency

equinox-testbed/Infrastructure.fs

@@ -16,10 +16,10 @@ module Guid =
 /// ClientId strongly typed id; represented internally as a Guid; not used for storage so rendering is not significant
 type ClientId = Guid<clientId>
 and [<Measure>] clientId
-module ClientId = let toStringN (value : ClientId) : string = Guid.toStringN %value
+module ClientId = let toString (value : ClientId) : string = Guid.toStringN %value
 
 /// SkuId strongly typed id; represented internally as a Guid
 // NB Perf is suboptimal as a key, see Equinox's samples/Store for expanded version
 type SkuId = Guid<skuId>
 and [<Measure>] skuId
-module SkuId = let toStringN (value : SkuId) : string = Guid.toStringN %value
+module SkuId = let toString (value : SkuId) : string = Guid.toStringN %value

equinox-testbed/Services.fs

@@ -7,6 +7,7 @@ module Domain =
 
  // NB - these schemas reflect the actual storage formats and hence need to be versioned with care
  module Events =
+
  type Favorited = { date: System.DateTimeOffset; skuId: SkuId }
  type Unfavorited = { skuId: SkuId }
  module Compaction =
@@ -20,6 +21,7 @@ module Domain =
  let codec = FsCodec.NewtonsoftJson.Codec.Create<Event>()
 
  module Folds =
+
  type State = Events.Favorited []
 
  type private InternalState(input: State) =
@@ -61,7 +63,8 @@ module Domain =
  [ Events.Unfavorited { skuId = skuId } ]
 
  type Service(log, resolveStream, ?maxAttempts) =
- let (|AggregateId|) (id: ClientId) = Equinox.AggregateId("Favorites", ClientId.toStringN id)
+
+ 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)

equinox-web/Domain/Aggregate.fs

@@ -2,13 +2,14 @@
 
 // NB - these types and names reflect the actual storage formats and hence need to be versioned with care
 module Events =
- type Compacted = { happened: bool }
+
+ type CompactedData = { happened: bool }
 
  type Event =
  | Happened
- | Compacted of Compacted
+ | Compacted of CompactedData
  interface TypeShape.UnionContract.IUnionContract
- let codec = FsCodec.NewtonsoftJson.Codec.Create<Event>()
+ let codec = FsCodec.NewtonsoftJson.Codec.Create<Event>(rejectNullaryCases=false)
 
 module Folds =
 
@@ -33,6 +34,7 @@ module Commands =
 type View = { sorted : bool }
 
 type Service(handlerLog, resolve, ?maxAttempts) =
+
  let (|AggregateId|) (id: string) = Equinox.AggregateId("Aggregate", id)
  let (|Stream|) (AggregateId id) = Equinox.Stream(handlerLog, resolve id, maxAttempts = defaultArg maxAttempts 2)
 

equinox-web/Domain/Infrastructure.fs

@@ -10,4 +10,4 @@ module Guid =
 type ClientId = Guid<clientId>
 and [<Measure>] clientId
 module ClientId =
- let toStringN (value : ClientId) : string = Guid.toStringN %value
+ let toString (value : ClientId) : string = Guid.toStringN %value

equinox-web/Domain/Todo.fs

@@ -72,8 +72,9 @@ type View = { id: int; order: int; title: string; completed: bool }
 
 /// Defines operations that a Controller can perform on a Todo List
 type Service(handlerLog, resolve, ?maxAttempts) =
+
  /// Maps a ClientId to the AggregateId that specifies the Stream in which the data for that client will be held
- let (|AggregateId|) (clientId: ClientId) = Equinox.AggregateId("Todos", ClientId.toStringN clientId)
+ let (|AggregateId|) (clientId: ClientId) = Equinox.AggregateId("Todos", ClientId.toString clientId)
 
  /// Maps a ClientId to Handler for the relevant stream
  let (|Stream|) (AggregateId id) = Equinox.Stream(handlerLog, resolve id, maxAttempts = defaultArg maxAttempts 2)

propulsion-consumer/Examples.fs

@@ -57,7 +57,7 @@ module MultiStreams =
  interface TypeShape.UnionContract.IUnionContract
  let codec = FsCodec.NewtonsoftJson.Codec.Create<Event>()
  let tryDecode = StreamCodec.tryDecode codec
- let [<Literal>] CategoryId = "SavedForLater"
+ let [<Literal>] categoryId = "SavedForLater"
 
  // NB - these schemas reflect the actual storage formats and hence need to be versioned with care
  module Favorites =
@@ -70,7 +70,7 @@ module MultiStreams =
  interface TypeShape.UnionContract.IUnionContract
  let codec = FsCodec.NewtonsoftJson.Codec.Create<Event>()
  let tryDecode = StreamCodec.tryDecode codec
- let [<Literal>] CategoryId = "Favorites"
+ let [<Literal>] categoryId = "Favorites"
 
  type Stat = Faves of int | Saves of int | OtherCategory of string * int | OtherMessage of string
 
@@ -85,10 +85,10 @@ module MultiStreams =
  let (|FavoritesEvents|SavedForLaterEvents|OtherCategory|UnknownMessage|) (streamName, span : Propulsion.Streams.StreamSpan<byte[]>) =
  let decode tryDecode = span.events |> Seq.choose (tryDecode log streamName) |> Array.ofSeq
  match category streamName with
- | Category (Favorites.CategoryId, id) ->
+ | Category (Favorites.categoryId, id) ->
  let s = match faves.TryGetValue id with true, value -> value | false, _ -> new HashSet<SkuId>()
  FavoritesEvents (id, s, decode Favorites.tryDecode)
- | Category (SavedForLater.CategoryId, id) ->
+ | Category (SavedForLater.categoryId, id) ->
  let s = match saves.TryGetValue id with true, value -> value | false, _ -> []
  SavedForLaterEvents (id, s, decode SavedForLater.tryDecode)
  | Category (categoryName, _) -> OtherCategory (categoryName, Seq.length span.events)
@@ -185,8 +185,8 @@ module MultiMessages =
  let span = Propulsion.Codec.NewtonsoftJson.RenderedSpan.Parse spanJson
  let decode tryDecode wrap = RenderedSpan.enum span |> Seq.choose (fun x -> x.event |> tryDecode log streamName |> Option.map wrap)
  match streamName with
- | Category (Favorites.CategoryId,_) -> yield! decode Favorites.tryDecode Fave
- | Category (SavedForLater.CategoryId,_) -> yield! decode SavedForLater.tryDecode Save
+ | Category (Favorites.categoryId,_) -> yield! decode Favorites.tryDecode Fave
+ | Category (SavedForLater.categoryId,_) -> yield! decode SavedForLater.tryDecode Save
  | Category (otherCategoryName,_) -> yield OtherCat (otherCategoryName, Seq.length span.e)
  | _ -> yield Unclassified streamName }
 

propulsion-consumer/Program.fs

@@ -73,8 +73,8 @@ let start (args : CmdParser.Arguments) =
 
 /// Handles command line parsing and running the program loop
 // NOTE Any custom logic should go in main
-let run argv =
- try use consumer = argv |> CmdParser.parse |> start
+let run args =
+ try use consumer = args |> CmdParser.parse |> start
  consumer.AwaitCompletion() |> Async.RunSynchronously
  if consumer.RanToCompletion then 0 else 2
  with :? Argu.ArguParseException as e -> eprintfn "%s" e.Message; 1

propulsion-consumer/Publisher.fs

@@ -36,7 +36,7 @@ module Input =
 
  let codec = FsCodec.NewtonsoftJson.Codec.Create<Event>()
  let tryDecode = StreamCodec.tryDecode codec
- let [<Literal>] CategoryId = "Inventory"
+ let [<Literal>] categoryId = "Inventory"
 
 module Output =
 
@@ -113,7 +113,7 @@ module Processor =
 
  let private enumStreamEvents(KeyValue (streamName : string, spanJson)) : seq<Propulsion.Streams.StreamEvent<_>> =
  match streamName with 
- | Category (Input.CategoryId,_) -> Propulsion.Codec.NewtonsoftJson.RenderedSpan.parse spanJson
+ | Category (Input.categoryId,_) -> Propulsion.Codec.NewtonsoftJson.RenderedSpan.parse spanJson
  | _ -> Seq.empty
 
  let log = Log.ForContext<Handler>()

propulsion-projector/Program.fs

@@ -213,8 +213,8 @@ let build (args : CmdParser.Arguments) =
 
 /// Handles command line parsing and running the program loop
 // NOTE Any custom logic should go in main
-let run argv =
- try let sink,runSourcePipeline = argv |> CmdParser.parse |> build
+let run args =
+ try let sink,runSourcePipeline = args |> CmdParser.parse |> build
  runSourcePipeline |> Async.Start
  sink.AwaitCompletion() |> Async.RunSynchronously
  if sink.RanToCompletion then 0 else 2

propulsion-summary-consumer/Infrastructure.fs

@@ -32,5 +32,5 @@ module Guid =
 type ClientId = Guid<clientId>
 and [<Measure>] clientId
 module ClientId =
- let toStringN (value : ClientId) : string = Guid.toStringN %value
+ let toString (value : ClientId) : string = Guid.toStringN %value
  let parse (value : string) : ClientId = let raw = Guid.Parse value in % raw

propulsion-summary-consumer/SumaryIngester.fs → propulsion-summary-consumer/Ingester.fs

@@ -1,6 +1,6 @@
 /// Follows a feed of updates, holding the most recently observed one; each update recieved is intended to completely supersede all previous updates
 /// Due to this, we should ensure that writes only happen where the update is not redundant and/or a replay of a previus message
-module ConsumerTemplate.SummaryIngester
+module ConsumerTemplate.Ingester
 
 open System
 
@@ -31,7 +31,7 @@ type Outcome = NoRelevantEvents of count : int | Ok of count : int | Skipped of
 type Stats(log, ?statsInterval, ?stateInterval) =
  inherit Propulsion.Kafka.StreamsConsumerStats<Outcome>(log, defaultArg statsInterval (TimeSpan.FromMinutes 1.), defaultArg stateInterval (TimeSpan.FromMinutes 5.))
 
- let mutable (ok, na, redundant) = 0, 0, 0
+ let mutable ok, na, redundant = 0, 0, 0
 
  override __.HandleOk res = res |> function
  | Outcome.Ok count -> ok <- ok + 1; redundant <- redundant + count - 1
@@ -51,7 +51,7 @@ let startConsumer (config : Jet.ConfluentKafka.FSharp.KafkaConsumerConfig) (log
  { items =
  [| for x in x.items ->
  { id = x.id; order = x.order; title = x.title; completed = x.completed } |]}
- let (|ClientId|) (value : string) = ClientId.parse value
+ let (|ClientId|) = ClientId.parse
  let (|DecodeNewest|_|) (codec : FsCodec.IUnionEncoder<_,_>) (stream, span : Propulsion.Streams.StreamSpan<_>) : 'summary option =
  span.events |> Seq.rev |> Seq.tryPick (StreamCodec.tryDecode codec log stream)
  let ingestIncomingSummaryMessage (stream, span : Propulsion.Streams.StreamSpan<_>) : Async<Outcome> = async {

propulsion-summary-consumer/Program.fs

@@ -112,12 +112,12 @@ let start (args : CmdParser.Arguments) =
  Jet.ConfluentKafka.FSharp.KafkaConsumerConfig.Create(
  appName, args.Broker, [args.Topic], args.Group,
  maxInFlightBytes = args.MaxInFlightBytes, ?statisticsInterval = args.LagFrequency)
- SummaryIngester.startConsumer config Log.Logger service args.MaxDop
+ Ingester.startConsumer config Log.Logger service args.MaxDop
 
 /// Handles command line parsing and running the program loop
 // NOTE Any custom logic should go in main
-let run argv =
- try use consumer = argv |> CmdParser.parse |> start
+let run args =
+ try use consumer = args |> CmdParser.parse |> start
  consumer.AwaitCompletion() |> Async.RunSynchronously
  if consumer.RanToCompletion then 0 else 2
  with :? Argu.ArguParseException as e -> eprintfn "%s" e.Message; 1

propulsion-summary-consumer/SummaryConsumer.fsproj

@@ -10,7 +10,7 @@
  <None Include="README.md" />
  <Compile Include="Infrastructure.fs" />
  <Compile Include="TodoSummary.fs" />
- <Compile Include="SumaryIngester.fs" />
+ <Compile Include="Ingester.fs" />
  <Compile Include="Program.fs" />
  </ItemGroup>
 

propulsion-summary-consumer/TodoSummary.fs

@@ -5,9 +5,9 @@ module Events =
 
  type ItemData = { id: int; order: int; title: string; completed: bool }
  type SummaryData = { items : ItemData[] }
-
+ type IngestedData = { version : int64; value : SummaryData }
  type Event =
- | Ingested of {| version: int64; value : SummaryData |}
+ | Ingested of IngestedData
  interface TypeShape.UnionContract.IUnionContract
  let codec = FsCodec.NewtonsoftJson.Codec.Create<Event>()
 
@@ -18,7 +18,7 @@ module Folds =
  let evolve _state = function
  | Events.Ingested e -> { version = e.version; value = Some e.value }
  let fold (state : State) : Events.Event seq -> State = Seq.fold evolve state
- let isOrigin = function _ -> true
+ let private isOrigin = function Events.Ingested _ -> true
  // A `transmute` function gets presented with:XX
  // a) events a command decided to generate (in it's `interpret`)
  // b) the state after applying them
@@ -31,18 +31,20 @@ module Folds =
  // and use AccessStrategy.RollingUnfolds with this `transmute` function so we instead convey:
  // a) "don't actually write these events we just decided on in `interpret` [and don't insert a new event batch document]"
  // b) "can you treat these events as snapshots please"
- let transmute events _state : Events.Event list * Events.Event list =
+ let private transmute events _state : Events.Event list * Events.Event list =
  [],events
+ // We don't want to write any events, so here we supply the `transmute` function to teach it how to treat our events as snapshots
+ let accessStrategy = Equinox.Cosmos.AccessStrategy.RollingUnfolds (isOrigin,transmute)
 
 module Commands =
  type Command =
- | Consume of version: int64 * Events.SummaryData
+ | Consume of version : int64 * value : Events.SummaryData
 
  let decide command (state : Folds.State) =
  match command with
  | Consume (version,value) ->
  if state.version <= version then false,[] else
- true,[Events.Ingested {| version = version; value = value |}]
+ true,[Events.Ingested { version = version; value = value }]
 
 type Item = { id: int; order: int; title: string; completed: bool }
 let render : Folds.State -> Item[] = function
@@ -54,11 +56,12 @@ let render : Folds.State -> Item[] = function
  completed = x.completed } |]
  | _ -> [||]
 
-let [<Literal>]categoryId = "TodoSummary"
+let [<Literal>] categoryId = "TodoSummary"
 
 /// Defines the operations that the Read side of a Controller and/or the Ingester can perform on the 'aggregate'
 type Service(log, resolve, ?maxAttempts) =
- let (|AggregateId|) (clientId: ClientId) = Equinox.AggregateId(categoryId, ClientId.toStringN clientId)
+
+ let (|AggregateId|) (clientId: ClientId) = Equinox.AggregateId(categoryId, ClientId.toString clientId)
  let (|Stream|) (AggregateId id) = Equinox.Stream<Events.Event,Folds.State>(log, resolve id, maxAttempts = defaultArg maxAttempts 2)
 
  let execute (Stream stream) command : Async<bool> =
@@ -74,10 +77,9 @@ type Service(log, resolve, ?maxAttempts) =
  query clientId render
 
 module Repository =
+
  open Equinox.Cosmos // Everything until now is independent of a concrete store
  let private resolve cache context =
- // We don't want to write any events, so here we supply the `transmute` function to teach it how to treat our events as snapshots
- let accessStrategy = AccessStrategy.RollingUnfolds (Folds.isOrigin, Folds.transmute)
  let cacheStrategy = CachingStrategy.SlidingWindow (cache, System.TimeSpan.FromMinutes 20.)
- Resolver(context, Events.codec, Folds.fold, Folds.initial, cacheStrategy, accessStrategy).Resolve
+ Resolver(context, Events.codec, Folds.fold, Folds.initial, cacheStrategy, Folds.accessStrategy).Resolve
  let createService cache context = Service(Serilog.Log.ForContext<Service>(), resolve cache context)

propulsion-summary-projector/Infrastructure.fs

@@ -27,7 +27,6 @@ module StreamNameParser =
  | [| category; id |] -> Category (category, id)
  | _ -> Unknown streamName
 
-
 module Guid =
  let inline toStringN (x : Guid) = x.ToString "N"
 

propulsion-summary-projector/Producer.fs

@@ -23,7 +23,7 @@ module Contract =
  let ofState (state : Todo.Folds.State) : SummaryEvent =
  Summary { items = [| for x in state.items -> render x |]}
 
-let (|ClientId|) (value : string) = ClientId.parse value
+let (|ClientId|) = ClientId.parse
 
 let (|Decode|) (codec : FsCodec.IUnionEncoder<_,_>) stream (span : Propulsion.Streams.StreamSpan<_>) =
  span.events |> Seq.choose (StreamCodec.tryDecodeSpan codec Serilog.Log.Logger stream)

propulsion-summary-projector/Program.fs

@@ -177,8 +177,8 @@ let build (args : CmdParser.Arguments) =
 
 /// Handles command line parsing and running the program loop
 // NOTE Any custom logic should go in main
-let run argv =
- try let projector,runSourcePipeline = argv |> CmdParser.parse |> build
+let run args =
+ try let projector,runSourcePipeline = args |> CmdParser.parse |> build
  runSourcePipeline |> Async.Start
  projector.AwaitCompletion() |> Async.RunSynchronously
  if projector.RanToCompletion then 0 else 2

propulsion-summary-projector/Todo.fs

@@ -14,7 +14,7 @@ module Events =
  | Cleared of {| nextId: int |}
  /// For EventStore, AccessStrategy.RollingSnapshots embeds these events every `batchSize` events
  /// For Cosmos, AccessStrategy.Snapshot maintains this as an event in the `u`nfolds list in the Tip-document
- | Snapshot of {| nextId: int; items: ItemData[] |}
+ | Snapshotted of {| nextId: int; items: ItemData[] |}
  interface TypeShape.UnionContract.IUnionContract
  let codec = FsCodec.NewtonsoftJson.Codec.Create<Event>()
 
@@ -31,20 +31,21 @@ module Folds =
  | Events.Updated value -> { s with items = s.items |> List.map (function { id = id } when id = value.id -> value | item -> item) }
  | Events.Deleted e -> { s with items = s.items |> List.filter (fun x -> x.id <> e.id) }
  | Events.Cleared e -> { nextId = e.nextId; items = [] }
- | Events.Snapshot s -> { nextId = s.nextId; items = List.ofArray s.items }
+ | Events.Snapshotted s -> { nextId = s.nextId; items = List.ofArray s.items }
  /// Folds a set of events from the store into a given `state`
  let fold (state : State) : Events.Event seq -> State = Seq.fold evolve state
  /// Determines whether a given event represents a checkpoint that implies we don't need to see any preceding events
- let isOrigin = function Events.Cleared _ | Events.Snapshot _ -> true | _ -> false
+ let isOrigin = function Events.Cleared _ | Events.Snapshotted _ -> true | _ -> false
  /// Prepares an Event that encodes all relevant aspects of a State such that `evolve` can rehydrate a complete State from it
- let snapshot state = Events.Snapshot {| nextId = state.nextId; items = Array.ofList state.items |}
- /// Allows us to slkip producing summaries for events that we know won't result in an externally discernable change to the summary output
- let impliesStateChange = function Events.Snapshot _ -> false | _ -> true
+ let snapshot state = Events.Snapshotted {| nextId = state.nextId; items = Array.ofList state.items |}
+ /// Allows us to skip producing summaries for events that we know won't result in an externally discernable change to the summary output
+ let impliesStateChange = function Events.Snapshotted _ -> false | _ -> true
 
-let [<Literal>]categoryId = "Todos"
+let [<Literal>] categoryId = "Todos"
 
 /// Defines operations that a Controller or Projector can perform on a Todo List
 type Service(log, resolve, ?maxAttempts) =
+
  /// Maps a ClientId to the AggregateId that specifies the Stream in which the data for that client will be held
  let (|AggregateId|) (clientId: ClientId) = Equinox.AggregateId(categoryId, ClientId.toString clientId)
 
@@ -60,6 +61,7 @@ type Service(log, resolve, ?maxAttempts) =
  queryEx clientId render
 
 module Repository =
+
  open Equinox.Cosmos // Everything until now is independent of a concrete store
  let private resolve cache context =
  let accessStrategy = AccessStrategy.Snapshot (Folds.isOrigin,Folds.snapshot)

propulsion-sync/Program.fs

@@ -575,8 +575,8 @@ let build (args : CmdParser.Arguments) =
 
 /// Handles command line parsing and running the program loop
 // NOTE Any custom logic should go in main
-let run argv =
- try let sink,runSourcePipeline = CmdParser.parse argv |> build
+let run args =
+ try let sink,runSourcePipeline = CmdParser.parse args |> build
  runSourcePipeline |> Async.Start
  sink.AwaitCompletion() |> Async.RunSynchronously
  if sink.RanToCompletion then 0 else 2