diff --git a/CHANGELOG.md b/CHANGELOG.md index abffcbc..6e2d3e6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -22,11 +22,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Breaking:** after `open FSharp.Control.R3`, the bare type name `Observable` refers to `FSharp.Control.R3.ObservableFactories.Observable`; write `open type R3.Observable` and extend `R3.Observable` by its full name - **Breaking:** `rxquery` `sumBy` adds values of the projected type: it requires `(+) : 'Value * 'Value -> 'Value` - `mapAsync` and `iterAsync` validate the concurrency limit of the options when they are called +- `chunkBySize` and `chunkBy` validate every chunk length when they are called and report the parameter and the rejected value ### Fixed - `rxquery` emitted its elements on the thread pool, out of order and after the source had moved on; `yield` and `zero` are now synchronous - `rxquery` `sumBy` passed `null` to the `(+)` of reference types +- `chunkBy` accepted a non-positive window length of `ChunkTimeSpanCount` and `ChunkMillisecondsCount`, which failed every element ## [0.3.1] - 2026-01-28 diff --git a/src/FSharp.Control.R3/Observable.fs b/src/FSharp.Control.R3/Observable.fs index 4d49ef3..c332b57 100644 --- a/src/FSharp.Control.R3/Observable.fs +++ b/src/FSharp.Control.R3/Observable.fs @@ -22,22 +22,43 @@ let inline catch ([] f : 'Exn -> Observable<'T>) o = ObservableE /// upn the successful termination of the first let inline concat first second = ObservableExtensions.Concat (first, second) -///Divides the input observable sequence into chunks of size at most chunkSize. -///The maximum size of each chunk. -///The input observable sequence. -///The observable sequence divided into chunks. -///Thrown when the input sequence is null. -///Thrown when chunkSize is not positive. -let inline chunkBySize (chunkSize : int) (source) = ObservableExtensions.Chunk (source, chunkSize) - -let inline chunkBy (configuration : ChunkConfiguration<'T>) (source) = +// R3 validates the length of Chunk(count) with the message as the parameter name and without the value, and does not validate +// the time-and-count overloads at all: there 0 makes every element fail with an IndexOutOfRangeException through OnErrorResume +// and a negative length fails only at subscription. The library validates every length itself, the same way. +let private ensurePositiveLength (paramName : string) (length : int) = + if length <= 0 then + raise (ArgumentOutOfRangeException (paramName, length, "The chunk length must be positive.")) + +/// Divides the source into chunks of at most elements. +/// The maximum size of each chunk. +/// The input observable sequence. +/// +/// The observable sequence divided into chunks. When the source completes, successfully or not, +/// the remaining elements are emitted as a last, shorter chunk. +/// +/// Thrown when is not positive. +let chunkBySize (chunkSize : int) (source : Observable<'T>) = + ensurePositiveLength (nameof chunkSize) chunkSize + ObservableExtensions.Chunk (source, chunkSize) + +/// +/// Divides the source into chunks as defined by . +/// When the source completes, the buffered elements are emitted as a last chunk. +/// +/// Thrown when the window length of the configuration is not positive. +let chunkBy (configuration : ChunkConfiguration<'T>) (source : Observable<'T>) = match configuration with - | ChunkCount count -> ObservableExtensions.Chunk (source, count) + | ChunkCount count -> + ensurePositiveLength (nameof configuration) count + ObservableExtensions.Chunk (source, count) | ChunkTimeSpan (timeSpan, timeProvider) -> ObservableExtensions.Chunk (source, timeSpan, timeProvider) - | ChunkTimeSpanCount (timeSpan, count, timeProvider) -> ObservableExtensions.Chunk (source, timeSpan, count, timeProvider) + | ChunkTimeSpanCount (timeSpan, count, timeProvider) -> + ensurePositiveLength (nameof configuration) count + ObservableExtensions.Chunk (source, timeSpan, count, timeProvider) | ChunkMilliseconds (milliseconds, timeProvider) -> ObservableExtensions.Chunk (source, TimeSpan.FromMilliseconds (float milliseconds), timeProvider) | ChunkMillisecondsCount (milliseconds, count, timeProvider) -> + ensurePositiveLength (nameof configuration) count ObservableExtensions.Chunk (source, TimeSpan.FromMilliseconds (float milliseconds), count, timeProvider) | ChunkAsyncWindow (asyncWindow, configureAwait) -> ObservableExtensions.Chunk (source, asyncWindow, configureAwait) | ChunkWindowBoundaries windowBoundaries -> ObservableExtensions.Chunk (source, windowBoundaries = windowBoundaries)