@@ -48,7 +48,7 @@ module Log =
4848// Sidestep Log.ForContext converting to a string; see https://github.com/serilog/serilog/issues/1124
4949 let event ( value : Event ) ( log : ILogger ) =
5050 let enrich ( e : LogEvent ) = e.AddPropertyIfAbsent( LogEventProperty( PropertyTag, ScalarValue( value)))
51- log.ForContext({ new Serilog.Core.ILogEventEnricher with member __ .( evt, _) = enrich evt })
51+ log.ForContext({ new Serilog.Core.ILogEventEnricher with member _ .( evt, _) = enrich evt })
5252 let withLoggedRetries < 't > retryPolicy ( contextLabel : string ) ( f : ILogger -> Async < 't >) log : Async < 't > =
5353 match retryPolicy with
5454 | None -> f log
@@ -83,9 +83,9 @@ module Log =
8383 type Counter =
8484 { mutable count : int64 ; mutable ms : int64 }
8585 static member Create () = { count = 0 L; ms = 0 L }
86- member __ .Ingest( ms ) =
87- System.Threading.Interlocked.Increment(&__ . count) |> ignore
88- System.Threading.Interlocked.Add(&__ . ms, ms) |> ignore
86+ member x .Ingest( ms ) =
87+ System.Threading.Interlocked.Increment(& x .count) |> ignore
88+ System.Threading.Interlocked.Add(& x .ms, ms) |> ignore
8989 type LogSink () =
9090 static let epoch = System.Diagnostics.Stopwatch.StartNew()
9191 static member val Read = Counter.Create() with get, set
@@ -99,7 +99,7 @@ module Log =
9999 epoch.Restart()
100100 span
101101 interface Serilog.Core.ILogEventSink with
102- member __ .Emit logEvent = logEvent |> function
102+ member _ .Emit logEvent = logEvent |> function
103103 | EsMetric ( Read stats) -> LogSink.Read.Ingest stats
104104 | EsMetric ( Write stats) -> LogSink.Write.Ingest stats
105105 | EsMetric ( Resync stats) -> LogSink.Resync.Ingest stats
@@ -339,34 +339,34 @@ module Token =
339339 newVersion > currentVersion
340340
341341type SqlStreamStoreConnection ( readConnection , [< O ; D ( null )>]? writeConnection , [< O ; D ( null )>]? readRetryPolicy , [< O ; D ( null )>]? writeRetryPolicy ) =
342- member __ .ReadConnection= readConnection
343- member __ .ReadRetryPolicy= readRetryPolicy
344- member __ .WriteConnection= defaultArg writeConnection readConnection
345- member __ .WriteRetryPolicy= writeRetryPolicy
342+ member _ .ReadConnection= readConnection
343+ member _ .ReadRetryPolicy= readRetryPolicy
344+ member _ .WriteConnection= defaultArg writeConnection readConnection
345+ member _ .WriteRetryPolicy= writeRetryPolicy
346346
347347type BatchingPolicy ( getMaxBatchSize : unit -> int , [< O ; D ( null )>]? batchCountLimit ) =
348348 new ( maxBatchSize ) = BatchingPolicy( fun () -> )
349- member __ .BatchSize= getMaxBatchSize()
350- member __ .MaxBatches= batchCountLimit
349+ member _ .BatchSize= getMaxBatchSize()
350+ member _ .MaxBatches= batchCountLimit
351351
352352[<RequireQualifiedAccess; NoComparison; NoEquality>]
353353type GatewaySyncResult = Written of StreamToken | ConflictUnknown
354354
355355type SqlStreamStoreContext ( connection : SqlStreamStoreConnection , batching : BatchingPolicy ) =
356- let isResolvedEventEventType ( tryDecode , predicate ) ( e : StreamMessage ) =
356+ let isResolvedEventEventType ( tryDecode , predicate ) ( e : StreamMessage ) =
357357 let data = e.GetJsonData() |> Async.AwaitTask |> Async.RunSynchronously
358358 predicate ( tryDecode data)
359359 let tryIsResolvedEventEventType predicateOption = predicateOption |> Option.map isResolvedEventEventType
360360 member internal _.LoadEmpty streamName = Token.ofUncompactedVersion batching.BatchSize streamName - 1 L
361- member _.LoadBatched streamName log ( tryDecode , isCompactionEventType ): Async < StreamToken * 'event []> = async {
361+ member _.LoadBatched streamName log ( tryDecode , isCompactionEventType ) : Async < StreamToken * 'event []> = async {
362362 let! version , events = Read.loadForwardsFrom log connection.ReadRetryPolicy connection.ReadConnection batching.BatchSize batching.MaxBatches streamName 0 L
363363 match tryIsResolvedEventEventType isCompactionEventType with
364364 | None -> return Token.ofNonCompacting streamName version, Array.choose tryDecode events
365365 | Some isCompactionEvent ->
366366 match events |> Array.tryFindBack isCompactionEvent with
367367 | None -> return Token.ofUncompactedVersion batching.BatchSize streamName version, Array.choose tryDecode events
368368 | Some resolvedEvent -> return Token.ofCompactionResolvedEventAndVersion resolvedEvent batching.BatchSize streamName version, Array.choose tryDecode events }
369- member _.LoadBackwardsStoppingAtCompactionEvent streamName log ( tryDecode , isOrigin ): Async < StreamToken * 'event []> = async {
369+ member _.LoadBackwardsStoppingAtCompactionEvent streamName log ( tryDecode , isOrigin ) : Async < StreamToken * 'event []> = async {
370370 let! version , events =
371371 Read.loadBackwardsUntilCompactionOrStart log connection.ReadRetryPolicy connection.ReadConnection batching.BatchSize batching.MaxBatches streamName ( tryDecode, isOrigin)
372372 match Array.tryHead events |> Option.filter ( function _, Some e -> isOrigin e | _ -> false ) with
@@ -400,7 +400,7 @@ type SqlStreamStoreContext(connection : SqlStreamStoreConnection, batching : Bat
400400 | Some compactionEventIndex ->
401401 Token.ofPreviousStreamVersionAndCompactionEventDataIndex streamToken compactionEventIndex encodedEvents.Length batching.BatchSize version'
402402 return GatewaySyncResult.Written token }
403- member __ .Sync( log , streamName , streamVersion , events : FsCodec.IEventData < byte []>[]) : Async < GatewaySyncResult > = async {
403+ member _ .Sync( log , streamName , streamVersion , events : FsCodec.IEventData < byte []>[]) : Async < GatewaySyncResult > = async {
404404 let encodedEvents : EventData [] = events |> Array.map UnionEncoderAdapters.eventDataOfEncodedEvent
405405 let! wr = Write.writeEvents log connection.WriteRetryPolicy connection.WriteConnection streamName streamVersion encodedEvents
406406 match wr with
@@ -425,7 +425,7 @@ type AccessStrategy<'event, 'state> =
425425
426426type private CompactionContext ( eventsLen : int , capacityBeforeCompaction : int ) =
427427 /// Determines whether writing a Compaction event is warranted (based on the existing state and the current accumulated changes)
428- member __ .IsCompactionDue= eventsLen > capacityBeforeCompaction
428+ member _ .IsCompactionDue= eventsLen > capacityBeforeCompaction
429429
430430type private Category < 'event , 'state , 'context >( context : SqlStreamStoreContext , codec : FsCodec.IEventCodec < _ , _ , 'context >, ? access : AccessStrategy < 'event , 'state >) =
431431 let tryDecode ( e : ResolvedEvent ) = e |> UnionEncoderAdapters.encodedEventOfResolvedEvent |> codec.TryDecode
@@ -453,13 +453,13 @@ type private Category<'event, 'state, 'context>(context : SqlStreamStoreContext,
453453 member _.LoadFromToken ( fold : 'state -> 'event seq -> 'state ) ( state : 'state ) ( streamName : string ) token ( log : ILogger ) : Async < StreamToken * 'state > =
454454 ( load fold) state ( context.LoadFromToken false streamName log token ( tryDecode, compactionPredicate))
455455 member _.TrySync < 'context >
456- ( log : ILogger , fold : 'state -> 'event seq -> 'state ,
456+ ( log : ILogger , fold : 'state -> 'event seq -> 'state ,
457457 ( Token.StreamPos ( stream , pos ) as streamToken ), state : 'state , events : 'event list , ctx : 'context option ): Async < SyncResult < 'state >> = async {
458- let encode e = codec.Encode( ctx, e)
458+ let encode e = codec.Encode( ctx, e)
459459 let events =
460460 match access with
461461 | None | Some AccessStrategy.LatestKnownEvent -> events
462- | Some ( AccessStrategy.RollingSnapshots (_, compact)) ->
462+ | Some ( AccessStrategy.RollingSnapshots (_, compact)) ->
463463 let cc = CompactionContext( List.length events, pos.batchCapacityLimit.Value)
464464 if cc.IsCompactionDue then events @ [ fold state events |> compact] else events
465465
@@ -595,7 +595,7 @@ type ConnectorBase([<O; D(null)>]?readRetryPolicy, [<O; D(null)>]?writeRetryPoli
595595
596596 abstract member Connect : unit -> Async < IStreamStore >
597597
598- member __ .Establish() : Async < SqlStreamStoreConnection > = async {
599- let! store = __ .()
598+ member x .Establish() : Async < SqlStreamStoreConnection > = async {
599+ let! store = x .Connect()
600600 return SqlStreamStoreConnection( readConnection= store, writeConnection= store, ?readRetryPolicy= readRetryPolicy, ?writeRetryPolicy= writeRetryPolicy)
601601 }
0 commit comments