Merge branch 'main' of https://github.com/fsprojects/FSharp.Control.AsyncSeq

Author: dsyme

.config/dotnet-tools.json

@@ -2,8 +2,8 @@
   "version": 1,
   "isRoot": true,
   "tools": {
-    "fsharp.formatting.commandtool": {
-      "version": "11.4.3",
+    "fsdocs-tool": {
+      "version": "21.0.0",
       "commands": [
         "fsdocs"
       ],

RELEASE_NOTES.md

@@ -1,3 +1,17 @@
+### 4.8.0
+
+* Added `AsyncSeq.mapFoldAsync` — maps each element using an asynchronous folder that also threads an accumulator state, returning both the array of results and the final state; mirrors `Seq.mapFold`.
+* Added `AsyncSeq.mapFold` — synchronous variant of `AsyncSeq.mapFoldAsync`, mirroring `Seq.mapFold`.
+* Added `AsyncSeq.allPairs` — returns an async sequence of all pairs from two input sequences (cartesian product); the second source is fully buffered before iteration, mirroring `Seq.allPairs`.
+* Added `AsyncSeq.rev` — returns a new async sequence with all elements in reverse order; the entire source sequence is buffered before yielding, mirroring `Seq.rev`.
+
+### 4.7.0
+
+* Added `AsyncSeq.splitAt` — splits a sequence at the given index, returning the first `count` elements as an array and the remaining elements as a new `AsyncSeq`. Mirrors `Seq.splitAt`. The source is enumerated once.
+* Added `AsyncSeq.removeAt` — returns a new sequence with the element at the specified index removed, mirroring `Seq.removeAt`.
+* Added `AsyncSeq.updateAt` — returns a new sequence with the element at the specified index replaced by a given value, mirroring `Seq.updateAt`.
+* Added `AsyncSeq.insertAt` — returns a new sequence with a value inserted before the element at the specified index (or appended if the index equals the sequence length), mirroring `Seq.insertAt`.
+
 ### 4.6.0
 
 * Added `AsyncSeq.isEmpty` — returns `true` if the sequence contains no elements; short-circuits after the first element, mirroring `Seq.isEmpty`.

docs/AsyncSeq.fsx

@@ -14,11 +14,11 @@ keywords: F#, asynchronous sequences, AsyncSeq, IAsyncEnumerable, async workflow
 #r "FSharp.Control.AsyncSeq.dll"
 (*** condition: fsx ***)
 #if FSX
-#r "nuget: FSharp.Control.AsyncSeq,{{package-version}}"
+#r "nuget: FSharp.Control.AsyncSeq,{{fsdocs-package-version}}"
 #endif // FSX
 (*** condition: ipynb ***)
 #if IPYNB
-#r "nuget: FSharp.Control.AsyncSeq,{{package-version}}"
+#r "nuget: FSharp.Control.AsyncSeq,{{fsdocs-package-version}}"
 #endif // IPYNB
 
 

docs/AsyncSeqExamples.fsx

@@ -14,11 +14,11 @@ keywords: F#, asynchronous sequences, AsyncSeq, examples
 #r "FSharp.Control.AsyncSeq.dll"
 (*** condition: fsx ***)
 #if FSX
-#r "nuget: FSharp.Control.AsyncSeq,{{package-version}}"
+#r "nuget: FSharp.Control.AsyncSeq,{{fsdocs-package-version}}"
 #endif // FSX
 (*** condition: ipynb ***)
 #if IPYNB
-#r "nuget: FSharp.Control.AsyncSeq,{{package-version}}"
+#r "nuget: FSharp.Control.AsyncSeq,{{fsdocs-package-version}}"
 #endif // IPYNB
 
 (**

docs/ComparisonWithObservable.fsx

@@ -14,11 +14,11 @@ keywords: F#, asynchronous sequences, AsyncSeq, IObservable, reactive
 #r "FSharp.Control.AsyncSeq.dll"
 (*** condition: fsx ***)
 #if FSX
-#r "nuget: FSharp.Control.AsyncSeq,{{package-version}}"
+#r "nuget: FSharp.Control.AsyncSeq,{{fsdocs-package-version}}"
 #endif // FSX
 (*** condition: ipynb ***)
 #if IPYNB
-#r "nuget: FSharp.Control.AsyncSeq,{{package-version}}"
+#r "nuget: FSharp.Control.AsyncSeq,{{fsdocs-package-version}}"
 #endif // IPYNB
 
 

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -1295,6 +1295,23 @@ module AsyncSeq =
   let reduce (f: 'T -> 'T -> 'T) (source: AsyncSeq<'T>) : Async<'T> =
       reduceAsync (fun a b -> f a b |> async.Return) source
 
+  let mapFoldAsync (folder: 'State -> 'T -> Async<'Result * 'State>) (state: 'State) (source: AsyncSeq<'T>) : Async<'Result array * 'State> = async {
+      let results = ResizeArray<'Result>()
+      let mutable st = state
+      use ie = source.GetEnumerator()
+      let! move = ie.MoveNext()
+      let b = ref move
+      while b.Value.IsSome do
+          let! (r, st') = folder st b.Value.Value
+          results.Add(r)
+          st <- st'
+          let! next = ie.MoveNext()
+          b := next
+      return (results.ToArray(), st) }
+
+  let mapFold (folder: 'State -> 'T -> 'Result * 'State) (state: 'State) (source: AsyncSeq<'T>) : Async<'Result array * 'State> =
+      mapFoldAsync (fun st x -> folder st x |> async.Return) state source
+
   let length (source : AsyncSeq<'T>) =
     fold (fun st _ -> st + 1L) 0L source
 
@@ -1432,6 +1449,32 @@ module AsyncSeq =
     let s = System.Collections.Generic.HashSet(excluded)
     source |> filter (fun x -> not (s.Contains(x)))
 
+  let removeAt (index : int) (source : AsyncSeq<'T>) : AsyncSeq<'T> = asyncSeq {
+    if index < 0 then invalidArg "index" "must be non-negative"
+    let i = ref 0
+    for x in source do
+      if i.Value <> index then yield x
+      i := i.Value + 1 }
+
+  let updateAt (index : int) (value : 'T) (source : AsyncSeq<'T>) : AsyncSeq<'T> = asyncSeq {
+    if index < 0 then invalidArg "index" "must be non-negative"
+    let i = ref 0
+    for x in source do
+      if i.Value = index then yield value
+      else yield x
+      i := i.Value + 1 }
+
+  let insertAt (index : int) (value : 'T) (source : AsyncSeq<'T>) : AsyncSeq<'T> = asyncSeq {
+    if index < 0 then invalidArg "index" "must be non-negative"
+    let i = ref 0
+    for x in source do
+      if i.Value = index then yield value
+      yield x
+      i := i.Value + 1
+    if i.Value = index then yield value
+    elif i.Value < index then
+      invalidArg "index" "The index is outside the range of elements in the collection." }
+
   #if !FABLE_COMPILER
   let iterAsyncParallel (f:'a -> Async<unit>) (s:AsyncSeq<'a>) : Async<unit> = async {
     use mb = MailboxProcessor.Start (ignore >> async.Return)
@@ -1676,6 +1719,25 @@ module AsyncSeq =
   let zipWith3 (f:'T1 -> 'T2 -> 'T3 -> 'U) (source1:AsyncSeq<'T1>) (source2:AsyncSeq<'T2>) (source3:AsyncSeq<'T3>) : AsyncSeq<'U> =
       zipWithAsync3 (fun a b c -> f a b c |> async.Return) source1 source2 source3
 
+  let allPairs (source1: AsyncSeq<'T1>) (source2: AsyncSeq<'T2>) : AsyncSeq<'T1 * 'T2> = asyncSeq {
+      let buf = System.Collections.Generic.List<'T2>()
+      use ie2 = source2.GetEnumerator()
+      let! move2 = ie2.MoveNext()
+      let b2 = ref move2
+      while b2.Value.IsSome do
+          buf.Add(b2.Value.Value)
+          let! next2 = ie2.MoveNext()
+          b2 := next2
+      use ie1 = source1.GetEnumerator()
+      let! move1 = ie1.MoveNext()
+      let b1 = ref move1
+      while b1.Value.IsSome do
+          let x = b1.Value.Value
+          for y in buf do
+              yield (x, y)
+          let! next1 = ie1.MoveNext()
+          b1 := next1 }
+
   let zappAsync (fs:AsyncSeq<'T -> Async<'U>>) (s:AsyncSeq<'T>) : AsyncSeq<'U> =
       zipWithAsync (|>) s fs
 
@@ -1815,6 +1877,35 @@ module AsyncSeq =
 
   let tail (source : AsyncSeq<'T>) : AsyncSeq<'T> = skip 1 source
 
+  /// Splits an async sequence at the given index, returning the first `count` elements as an array
+  /// and the remaining elements as a new AsyncSeq. The source is enumerated once.
+  let splitAt (count: int) (source: AsyncSeq<'T>) : Async<'T array * AsyncSeq<'T>> = async {
+      if count < 0 then invalidArg "count" "must be non-negative"
+      let ie = source.GetEnumerator()
+      let ra = ResizeArray<'T>()
+      let! m = ie.MoveNext()
+      let b = ref m
+      while b.Value.IsSome && ra.Count < count do
+          ra.Add b.Value.Value
+          let! next = ie.MoveNext()
+          b := next
+      let first = ra.ToArray()
+      let rest =
+          if b.Value.IsNone then
+              ie.Dispose()
+              empty<'T>
+          else
+              let cur = ref b.Value
+              asyncSeq {
+                  try
+                      while cur.Value.IsSome do
+                          yield cur.Value.Value
+                          let! next = ie.MoveNext()
+                          cur := next
+                  finally
+                      ie.Dispose() }
+      return first, rest }
+
   let toArrayAsync (source : AsyncSeq<'T>) : Async<'T[]> = async {
       let ra = (new ResizeArray<_>())
       use ie = source.GetEnumerator()
@@ -1939,6 +2030,11 @@ module AsyncSeq =
 
   let sortWith (comparer:'T -> 'T -> int) (source:AsyncSeq<'T>) : array<'T> =
     toSortedSeq (Array.sortWith comparer) source
+
+  let rev (source: AsyncSeq<'T>) : AsyncSeq<'T> = asyncSeq {
+      let! arr = toArrayAsync source
+      for i in arr.Length - 1 .. -1 .. 0 do
+          yield arr.[i] }
   #endif
 
   #if !FABLE_COMPILER

src/FSharp.Control.AsyncSeq/AsyncSeq.fsi

@@ -258,6 +258,16 @@ module AsyncSeq =
     /// specified 'reduction' function. Raises InvalidOperationException if the sequence is empty.
     val reduce : reduction:('T -> 'T -> 'T) -> source:AsyncSeq<'T> -> Async<'T>
 
+    /// Asynchronously maps each element of the async sequence with an asynchronous folder function that
+    /// also threads an accumulator state through the computation. Returns the array of results and the
+    /// final state, mirroring Seq.mapFold.
+    val mapFoldAsync : folder:('State -> 'T -> Async<'Result * 'State>) -> state:'State -> source:AsyncSeq<'T> -> Async<'Result array * 'State>
+
+    /// Maps each element of the async sequence with a folder function that also threads an accumulator
+    /// state through the computation. Returns the array of results and the final state,
+    /// mirroring Seq.mapFold.
+    val mapFold : folder:('State -> 'T -> 'Result * 'State) -> state:'State -> source:AsyncSeq<'T> -> Async<'Result array * 'State>
+
     /// Asynchronously sum the elements of the input asynchronous sequence using the specified function.
     val inline sum : source:AsyncSeq< ^T > -> Async< ^T>
                                 when ^T : (static member ( + ) : ^T * ^T -> ^T)
@@ -408,6 +418,19 @@ module AsyncSeq =
     /// in the given excluded collection. Uses a HashSet for O(1) lookup. Mirrors Seq.except.
     val except : excluded:seq<'T> -> source:AsyncSeq<'T> -> AsyncSeq<'T> when 'T : equality
 
+    /// Returns a new asynchronous sequence with the element at the specified index removed.
+    /// Raises ArgumentException if index is negative. Mirrors Seq.removeAt.
+    val removeAt : index:int -> source:AsyncSeq<'T> -> AsyncSeq<'T>
+
+    /// Returns a new asynchronous sequence with the element at the specified index replaced by the given value.
+    /// Raises ArgumentException if index is negative. Mirrors Seq.updateAt.
+    val updateAt : index:int -> value:'T -> source:AsyncSeq<'T> -> AsyncSeq<'T>
+
+    /// Returns a new asynchronous sequence with the given value inserted before the element at the specified index.
+    /// An index equal to the length of the sequence appends the value at the end.
+    /// Raises ArgumentException if index is negative or greater than the sequence length. Mirrors Seq.insertAt.
+    val insertAt : index:int -> value:'T -> source:AsyncSeq<'T> -> AsyncSeq<'T>
+
     /// Creates an asynchronous sequence that lazily takes element from an
     /// input synchronous sequence and returns them one-by-one.
     val ofSeq : source:seq<'T> -> AsyncSeq<'T>
@@ -483,6 +506,10 @@ module AsyncSeq =
     /// The resulting sequence stops when any of the argument sequences stop.
     val zipWith3 : mapping:('T1 -> 'T2 -> 'T3 -> 'U) -> source1:AsyncSeq<'T1> -> source2:AsyncSeq<'T2> -> source3:AsyncSeq<'T3> -> AsyncSeq<'U>
 
+    /// Returns an async sequence of all pairs of elements from the two input sequences.
+    /// The second sequence is fully buffered before iteration begins, mirroring Seq.allPairs.
+    val allPairs : source1:AsyncSeq<'T1> -> source2:AsyncSeq<'T2> -> AsyncSeq<'T1 * 'T2>
+
     /// Builds a new asynchronous sequence whose elements are generated by
     /// applying the specified function to all elements of the input sequence.
     ///
@@ -586,6 +613,11 @@ module AsyncSeq =
     /// Returns an empty sequence if the source is empty.
     val tail : source:AsyncSeq<'T> -> AsyncSeq<'T>
 
+    /// Splits an async sequence at the given index. Returns an async computation that yields
+    /// the first `count` elements as an array and the remaining elements as a new AsyncSeq.
+    /// The source is enumerated once; the returned AsyncSeq lazily produces the remainder.
+    val splitAt : count:int -> source:AsyncSeq<'T> -> Async<'T array * AsyncSeq<'T>>
+
     /// Creates an async computation which iterates the AsyncSeq and collects the output into an array.
     val toArrayAsync : source:AsyncSeq<'T> -> Async<'T []>
 
@@ -636,6 +668,11 @@ module AsyncSeq =
     /// that sequence is iterated. As a result this function should not be used with
     /// large or infinite sequences.
     val sortWith : comparer:('T -> 'T -> int) -> source:AsyncSeq<'T> -> array<'T>
+
+    /// Returns a new async sequence with the elements in reverse order. The entire source
+    /// sequence is buffered before yielding any elements, mirroring Seq.rev.
+    /// This function should not be used with large or infinite sequences.
+    val rev : source:AsyncSeq<'T> -> AsyncSeq<'T>
     #endif
 
     /// Interleaves two async sequences of the same type into a resulting sequence. The provided

src/FSharp.Control.AsyncSeq/FSharp.Control.AsyncSeq.fsproj

@@ -9,6 +9,8 @@
     <PackageProjectUrl>https://fsprojects.github.io/FSharp.Control.AsyncSeq/</PackageProjectUrl>
     <PackageIconUrl>https://fsprojects.github.io/FSharp.Control.AsyncSeq/img/logo.png</PackageIconUrl>
     <PackageTags>F#;async;fsharp;streaming</PackageTags>
+    <FsDocsLicenseLink>https://github.com/fsprojects/FSharp.Control.AsyncSeq/blob/main/LICENSE.md</FsDocsLicenseLink>
+    <FsDocsReleaseNotesLink>https://github.com/fsprojects/FSharp.Control.AsyncSeq/blob/main/RELEASE_NOTES.md</FsDocsReleaseNotesLink>
     <IncludeSymbols>true</IncludeSymbols>
     <RepositoryUrl>https://github.com/fsprojects/FSharp.Control.AsyncSeq</RepositoryUrl>
     <SymbolPackageFormat>snupkg</SymbolPackageFormat>