diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index d1a67bc..11e07bf 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -14,7 +14,7 @@ / ├── src/FSharp.Control.R3/ – main library │ ├── AssemblyInfo.fs – assembly metadata -│ ├── ProcessingOptions.fs – processing configuration helpers +│ ├── ProcessingOptions.fs – processing and chunk configuration │ ├── Observable.fs – observable wrappers/extensions │ ├── ObservableOption.fs – `option` variants of the `Observable` functions (`choose`) │ ├── ObservableFactories.fs – factories reachable as `Observable.xxx` (`ofSeq`) diff --git a/src/FSharp.Control.R3/ProcessingOptions.fs b/src/FSharp.Control.R3/ProcessingOptions.fs index cd4ca69..9a3a340 100644 --- a/src/FSharp.Control.R3/ProcessingOptions.fs +++ b/src/FSharp.Control.R3/ProcessingOptions.fs @@ -19,20 +19,28 @@ type AwaitOperationConfiguration = | AwaitDrop /// If the previous asynchronous method is running, it is cancelled and the next asynchronous method is executed. | AwaitSwitch - /// All values are sent immediately to the asynchronous method. + /// Up to MaxConcurrent values, or all of them when it is -1, are sent to the asynchronous method at once; the others wait in a queue. | AwaitParallel of /// Maximum number of concurrent invocations; -1 means no limit, otherwise it must be greater than 0. MaxConcurrent : int - /// All values are sent immediately to the asynchronous method, but the results are queued and passed to the next operator in order. + /// + /// Up to MaxConcurrent values, or all of them when it is -1, are sent to the asynchronous method at once and the others wait in a queue, + /// but the results are passed to the next operator in the order of the values. + /// | AwaitSequentialParallel of /// Maximum number of concurrent invocations; -1 means no limit, otherwise it must be greater than 0. MaxConcurrent : int /// Send the first value and the last value while the asynchronous method is running. | AwaitThrottleFirstLast +/// Options that configure how an asynchronous selector is applied to an observable sequence. type ProcessingOptions = { + /// Defines how elements that arrive while a previous invocation is running are processed. AwaitOperationConfiguration : AwaitOperationConfiguration + /// Whether R3 resumes on the captured synchronization context after awaiting the result of the asynchronous selector. + /// The awaits inside the selector capture the context on their own. ConfigureAwait : bool + /// Whether the running invocations are cancelled, their results dropped and the queued elements discarded when the source completes successfully. CancelOnCompleted : bool } with @@ -48,9 +56,13 @@ type ProcessingOptions = { CancelOnCompleted = false } + /// Processes the elements one at a time in arrival order, capturing the synchronization context. static member Default = ``default`` + + /// Processes all elements concurrently without a limit, capturing the synchronization context. static member Parallel = ``parallel`` + /// The concurrency limit of the parallel configurations; -1 (no limit) for every other configuration. member this.MaxConcurrent = match this.AwaitOperationConfiguration with | AwaitOperationConfiguration.AwaitSequential -> -1 @@ -60,6 +72,7 @@ type ProcessingOptions = { | AwaitOperationConfiguration.AwaitSequentialParallel maxConcurrent -> maxConcurrent | AwaitOperationConfiguration.AwaitThrottleFirstLast -> -1 + /// The R3 that corresponds to . member this.AwaitOperation = match this.AwaitOperationConfiguration with | AwaitOperationConfiguration.AwaitSequential -> AwaitOperation.Sequential @@ -83,22 +96,91 @@ type ProcessingOptions = { raise (ArgumentOutOfRangeException (paramName, maxConcurrent, "MaxConcurrent must be -1 (no limit) or greater than 0.")) | _ -> () +/// Defines how an observable sequence is divided into chunks. type ChunkConfiguration<'T> = - | ChunkCount of WindowLength : int - | ChunkTimeSpan of WindowTime : TimeSpan * TimeProvider : TimeProvider - | ChunkTimeSpanCount of WindowTime : TimeSpan * WindowLength : int * TimeProvider : TimeProvider - | ChunkMilliseconds of WindowTime : int * TimeProvider : TimeProvider - | ChunkMillisecondsCount of WindowTime : int * WindowLength : int * TimeProvider : TimeProvider - | ChunkAsyncWindow of AsyncWindow : Func<'T, CancellationToken, ValueTask> * ConfigureAwait : bool - | ChunkWindowBoundaries of WindowBoundaries : Observable<'T> + /// Chunks of at most a number of elements. + | ChunkCount of + /// Maximum number of elements of a chunk. + WindowLength : int + /// Chunks of the elements received during a time span, measured from the first element of every chunk. + | ChunkTimeSpan of + /// Duration of a chunk, measured from its first element. + WindowTime : TimeSpan * + /// Time provider that measures the duration. + TimeProvider : TimeProvider + /// Chunks of at most a number of elements received during a time span, measured from the first element of every chunk. + | ChunkTimeSpanCount of + /// Duration of a chunk, measured from its first element. + WindowTime : TimeSpan * + /// Maximum number of elements of a chunk. + WindowLength : int * + /// Time provider that measures the duration. + TimeProvider : TimeProvider + /// Chunks of the elements received during a number of milliseconds, measured from the first element of every chunk. + | ChunkMilliseconds of + /// Duration of a chunk in milliseconds, measured from its first element. + WindowTime : int * + /// Time provider that measures the duration. + TimeProvider : TimeProvider + /// Chunks of at most a number of elements received during a number of milliseconds, measured from the first element of every chunk. + | ChunkMillisecondsCount of + /// Duration of a chunk in milliseconds, measured from its first element. + WindowTime : int * + /// Maximum number of elements of a chunk. + WindowLength : int * + /// Time provider that measures the duration. + TimeProvider : TimeProvider + /// Chunks that start with an element and end when the asynchronous window started for that element completes. + | ChunkAsyncWindow of + /// Starts the window of a chunk for its first element; the chunk ends when the returned task completes. + AsyncWindow : Func<'T, CancellationToken, ValueTask> * + /// Whether the continuation after the window resumes on the captured synchronization context. + ConfigureAwait : bool + /// + /// Chunks that end on every element of a boundary sequence. + /// + /// The boundaries must have the element type of the source; + /// + /// accepts boundaries of any element type. + /// + /// + | ChunkWindowBoundaries of + /// Every element of this sequence ends a chunk. + WindowBoundaries : Observable<'T> +/// +/// Helpers that create values. +/// +/// The time based helpers read when they are called; +/// use the union cases directly to pass another time provider. +/// +/// module ChunkConfiguration = + + /// Chunks of the elements received during , measured by the default time provider. let inline TimeSpan windowTime = ChunkTimeSpan (windowTime, ObservableSystem.DefaultTimeProvider) + + /// + /// Chunks of at most elements received during , + /// measured by the default time provider. + /// let inline TimeSpanCount windowTime windowLength = ChunkTimeSpanCount (windowTime, windowLength, ObservableSystem.DefaultTimeProvider) + + /// Chunks of the elements received during milliseconds, measured by the default time provider. let inline Milliseconds windowTime = ChunkMilliseconds (windowTime, ObservableSystem.DefaultTimeProvider) + + /// + /// Chunks of at most elements received during milliseconds, + /// measured by the default time provider. + /// let inline MillisecondsCount windowTime windowLength = ChunkMillisecondsCount (windowTime, windowLength, ObservableSystem.DefaultTimeProvider) + + /// + /// Chunks that start with an element and end when the computation returns for that element completes. + /// The computation runs with a cancellation token that is cancelled when the chunked sequence completes or is disposed. + /// let AsyncWindow (asyncWindow : 'T -> Async) = let asyncWindow element ct = Async.StartImmediateAsTask (asyncWindow element, ct) :> Task