CosmosStoreSource: Expose tailSleepInterval

also make startFromTail optional
jet · Sep 7, 2022 · f5208b9 · f5208b9
1 parent c3b0e78
commit f5208b9
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 3 deletions.
diff --git a/src/Propulsion.CosmosStore/CosmosStoreSource.fs b/src/Propulsion.CosmosStore/CosmosStoreSource.fs
@@ -132,7 +132,7 @@ type CosmosStoreSource =
     static member Start
         (   log : ILogger,
             monitored : Container, leases : Container, processorName, observer,
-            startFromTail, ?maxItems, ?lagReportFreq : TimeSpan, ?notifyError, ?customize) =
+            ?maxItems, ?tailSleepInterval, ?startFromTail, ?lagReportFreq : TimeSpan, ?notifyError, ?customize) =
         let databaseId, containerId = monitored.Database.Id, monitored.Id
 #endif
         let logLag (interval : TimeSpan) (remainingWork : (int*int64) list) = async {
@@ -158,7 +158,8 @@ type CosmosStoreSource =
         let source =
             ChangeFeedProcessor.Start
               ( log, monitored, leases, processorName, observer, ?notifyError=notifyError, ?customize=customize,
-                startFromTail=startFromTail, ?reportLagAndAwaitNextEstimation=maybeLogLag, ?maxItems=maxItems,
+                ?maxItems = maxItems, ?feedPollDelay = tailSleepInterval, ?reportLagAndAwaitNextEstimation=maybeLogLag,
+                startFromTail = defaultArg startFromTail false,
                 leaseAcquireInterval=TimeSpan.FromSeconds 5., leaseRenewInterval=TimeSpan.FromSeconds 5., leaseTtl=TimeSpan.FromSeconds 10.)
         lagReportFreq |> Option.iter (fun s -> log.Information("ChangeFeed {processorName} Lag stats interval {lagReportIntervalS:n0}s", processorName, s.TotalSeconds))
         source

diff --git a/src/Propulsion.SqlStreamStore/SqlStreamStoreSource.fs b/src/Propulsion.SqlStreamStore/SqlStreamStoreSource.fs
@@ -45,7 +45,7 @@ type SqlStreamStoreSource
     (   log : Serilog.ILogger, statsInterval,
         store : SqlStreamStore.IStreamStore, batchSize, tailSleepInterval,
         checkpoints : Propulsion.Feed.IFeedCheckpointStore, sink : Propulsion.Streams.Default.Sink,
-        ?categoryFilter : string -> bool,
+        categoryFilter : string -> bool,
         // If the Handler does not require the Data/Meta of the events, the query to load the events can be much more efficient. Default: false
         ?hydrateBodies,
         ?startFromTail,