diff --git a/CHANGELOG.md b/CHANGELOG.md index 20f2f3a..4dff414 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `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 - A positional comparer or element selector passed to the `Async` `toLookup` was silently ignored +- `iterAsync` could fault with `OverflowException` on sources with more than `Int32.MaxValue` elements - `chunkBy` accepted a non-positive window length of `ChunkTimeSpanCount` and `ChunkMillisecondsCount`, which failed every element - Misleading XML docs of `catch` diff --git a/src/FSharp.Control.R3/AsyncObservable.fs b/src/FSharp.Control.R3/AsyncObservable.fs index e29717b..3f5f256 100644 --- a/src/FSharp.Control.R3/AsyncObservable.fs +++ b/src/FSharp.Control.R3/AsyncObservable.fs @@ -144,7 +144,9 @@ module Observable = /// by intercepting the message stream to run arbitrary actions for messages on the pipeline. /// /// Thrown when the concurrency limit of the options is 0 or below -1. - let iterAsync options (action : 't -> Async) source = source |> mapAsync options action |> length |> Async.Ignore + let iterAsync options (action : 't -> Async) source = + // Waits through iter: waiting through length counted the elements with a checked add, which overflows on long-lived sources + source |> mapAsync options action |> iter ignore [] module Extensions = diff --git a/src/FSharp.Control.R3/TaskObservable.fs b/src/FSharp.Control.R3/TaskObservable.fs index 60a28b3..b8237c3 100644 --- a/src/FSharp.Control.R3/TaskObservable.fs +++ b/src/FSharp.Control.R3/TaskObservable.fs @@ -61,10 +61,10 @@ module Observable = /// /// Thrown when the concurrency limit of the options is 0 or below -1. let iterAsync cancellationToken options (action : CancellationToken -> 't -> Task) source = + // Waits through iter: waiting through length counted the elements with a checked add, which overflows on long-lived sources source |> mapAsync options action - |> length cancellationToken - :> Task + |> iter cancellationToken ignore [] module Extensions =