diff --git a/src/Equinox/Decider.fs b/src/Equinox/Decider.fs index 73e957f50..98459da33 100755 --- a/src/Equinox/Decider.fs +++ b/src/Equinox/Decider.fs @@ -44,12 +44,12 @@ type Decider<'event, 'state> /// 1b. Tries up to maxAttempts times in the case of a conflict, throwing MaxResyncsExhaustedException to signal failure. /// 2. Uses mapResult to render the final outcome from the 'result and/or the final ISyncContext /// 3. Yields the outcome - member __.TransactEx(decide : ISyncContext<'state> -> Async<'result * 'event list>, mapResult : 'result -> ISyncContext<'state> -> 'resultEx) : Async<'resultEx> = + member __.TransactEx(decide : ISyncContext<'state> -> Async<'result * 'event list>, mapResult : 'result -> ISyncContext<'state> -> 'view) : Async<'view> = transact decide mapResult /// Project from the folded 'state, without executing a decision flow as Transact does member __.Query(projection : 'state -> 'view) : Async<'view> = - Flow.query (stream, log, fun syncState -> projection (syncState :> ISyncContext<'state>).State) + Flow.query (stream, log, fun context -> projection (context :> ISyncContext<'state>).State) /// Project from the stream's 'state (including extended context), without executing a decision flow as Transact does member __.QueryEx(projection : ISyncContext<'state> -> 'view) : Async<'view> = diff --git a/src/Equinox/Flow.fs b/src/Equinox/Flow.fs index 223f5b8b0..0dd61c55e 100755 --- a/src/Equinox/Flow.fs +++ b/src/Equinox/Flow.fs @@ -44,7 +44,7 @@ type ISyncContext<'state> = module internal Flow = /// Represents stream and folding state between the load and run/render phases - type SyncState<'event, 'state> + type SyncContext<'event, 'state> ( originState : StreamToken * 'state, trySync : ILogger * StreamToken * 'state * 'event list -> Async>) = let mutable tokenAndState = originState @@ -78,45 +78,45 @@ module internal Flow = /// 2b. if saved without conflict, exit with updated state /// 2b. if conflicting changes, retry by recommencing at step 1 with the updated state let run (log : ILogger) (maxSyncAttempts : int, resyncRetryPolicy, createMaxAttemptsExhaustedException) - (syncState : SyncState<'event, 'state>) + (context : SyncContext<'event, 'state>) (decide : ISyncContext<'state> -> Async<'result * 'event list>) - (mapResult : 'result -> SyncState<'event, 'state> -> 'resultEx) - : Async<'resultEx> = + (mapResult : 'result -> SyncContext<'event, 'state> -> 'view) + : Async<'view> = if maxSyncAttempts < 1 then raise <| System.ArgumentOutOfRangeException("maxSyncAttempts", maxSyncAttempts, "should be >= 1") /// Run a decision cycle - decide what events should be appended given the presented state - let rec loop attempt : Async<'resultEx> = async { + let rec loop attempt : Async<'view> = async { let log = if attempt = 1 then log else log.ForContext("syncAttempt", attempt) - let! result, events = decide (syncState :> ISyncContext<'state>) + let! result, events = decide (context :> ISyncContext<'state>) if List.isEmpty events then log.Debug "No events generated" - return mapResult result syncState + return mapResult result context elif attempt = maxSyncAttempts then // Special case: on final attempt, we won't be `resync`ing; we're giving up - let! committed = syncState.TryWithoutResync(log, events) + let! committed = context.TryWithoutResync(log, events) if not committed then log.Debug "Max Sync Attempts exceeded" return raise (createMaxAttemptsExhaustedException attempt) else - return mapResult result syncState + return mapResult result context else - let! committed = syncState.TryOrResync(resyncRetryPolicy, attempt, log, events) + let! committed = context.TryOrResync(resyncRetryPolicy, attempt, log, events) if not committed then log.Debug "Resyncing and retrying" return! loop (attempt + 1) else - return mapResult result syncState } + return mapResult result context } // Commence, processing based on the incoming state loop 1 let transact (maxAttempts, resyncRetryPolicy, createMaxAttemptsExhaustedException) (stream : IStream<_, _>, log) decide mapResult : Async<'result> = async { let! streamState = stream.Load log - let syncState = SyncState(streamState, stream.TrySync) - return! run log (maxAttempts, resyncRetryPolicy, createMaxAttemptsExhaustedException) syncState decide mapResult } + let context = SyncContext(streamState, stream.TrySync) + return! run log (maxAttempts, resyncRetryPolicy, createMaxAttemptsExhaustedException) context decide mapResult } - let query (stream : IStream<'event, 'state>, log : ILogger, project: SyncState<'event, 'state> -> 'result) : Async<'result> = async { + let query (stream : IStream<'event, 'state>, log : ILogger, project: SyncContext<'event, 'state> -> 'result) : Async<'result> = async { let! streamState = stream.Load log - let syncState = SyncState(streamState, stream.TrySync) - return project syncState } + let context = SyncContext(streamState, stream.TrySync) + return project context }