Merge branch 'main' into repo-assist/add-contributing-guidelines-c9f0ce702280ab97

Author: dsyme

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -1087,6 +1087,21 @@ module AsyncSeq =
           let! moven = ie.MoveNext()
           b := moven }
 
+  let windowed (windowSize:int) (source:AsyncSeq<'T>) : AsyncSeq<'T[]> =
+    if windowSize < 1 then invalidArg (nameof windowSize) "must be positive"
+    asyncSeq {
+      let window = System.Collections.Generic.Queue<'T>(windowSize)
+      use ie = source.GetEnumerator()
+      let! move = ie.MoveNext()
+      let b = ref move
+      while b.Value.IsSome do
+          window.Enqueue(b.Value.Value)
+          if window.Count = windowSize then
+              yield window.ToArray()
+              window.Dequeue() |> ignore
+          let! moven = ie.MoveNext()
+          b := moven }
+
   let pickAsync (f:'T -> Async<'U option>) (source:AsyncSeq<'T>) = async {
       use ie = source.GetEnumerator()
       let! v = ie.MoveNext()
@@ -1145,12 +1160,58 @@ module AsyncSeq =
   let fold f (state:'State) (source : AsyncSeq<'T>) =
     foldAsync (fun st v -> f st v |> async.Return) state source
 
+  let reduceAsync (f: 'T -> 'T -> Async<'T>) (source: AsyncSeq<'T>) : Async<'T> = async {
+      use ie = source.GetEnumerator()
+      let! first = ie.MoveNext()
+      match first with
+      | None -> return raise (InvalidOperationException("The input sequence was empty."))
+      | Some v ->
+          let acc = ref v
+          let! next = ie.MoveNext()
+          let b = ref next
+          while b.Value.IsSome do
+              let! newAcc = f acc.Value b.Value.Value
+              acc := newAcc
+              let! more = ie.MoveNext()
+              b := more
+          return acc.Value }
+
+  let reduce (f: 'T -> 'T -> 'T) (source: AsyncSeq<'T>) : Async<'T> =
+      reduceAsync (fun a b -> f a b |> async.Return) source
+
   let length (source : AsyncSeq<'T>) =
     fold (fun st _ -> st + 1L) 0L source
 
   let inline sum (source : AsyncSeq<'T>) : Async<'T> =
     (LanguagePrimitives.GenericZero, source) ||> fold (+)
 
+  let inline sumBy (projection : 'T -> ^U) (source : AsyncSeq<'T>) : Async<^U> =
+    fold (fun s x -> s + projection x) LanguagePrimitives.GenericZero source
+
+  let inline sumByAsync (projection : 'T -> Async<^U>) (source : AsyncSeq<'T>) : Async<^U> =
+    foldAsync (fun s x -> async { let! v = projection x in return s + v }) LanguagePrimitives.GenericZero source
+
+  let inline average (source : AsyncSeq<^T>) : Async<^T> =
+    async {
+      let! sum, count = fold (fun (s, n) x -> (s + x, n + 1)) (LanguagePrimitives.GenericZero, 0) source
+      if count = 0 then invalidArg "source" "The input sequence was empty."
+      return LanguagePrimitives.DivideByInt sum count
+    }
+
+  let inline averageBy (projection : 'T -> ^U) (source : AsyncSeq<'T>) : Async<^U> =
+    async {
+      let! sum, count = fold (fun (s, n) x -> (s + projection x, n + 1)) (LanguagePrimitives.GenericZero, 0) source
+      if count = 0 then invalidArg "source" "The input sequence was empty."
+      return LanguagePrimitives.DivideByInt sum count
+    }
+
+  let inline averageByAsync (projection : 'T -> Async<^U>) (source : AsyncSeq<'T>) : Async<^U> =
+    async {
+      let! sum, count = foldAsync (fun (s, n) x -> async { let! v = projection x in return (s + v, n + 1) }) (LanguagePrimitives.GenericZero, 0) source
+      if count = 0 then invalidArg "source" "The input sequence was empty."
+      return LanguagePrimitives.DivideByInt sum count
+    }
+
   let scan f (state:'State) (source : AsyncSeq<'T>) =
     scanAsync (fun st v -> f st v |> async.Return) state source
 

src/FSharp.Control.AsyncSeq/AsyncSeq.fsi

@@ -198,6 +198,14 @@ module AsyncSeq =
     /// singleton input sequence.
     val pairwise : source:AsyncSeq<'T> -> AsyncSeq<'T * 'T>
 
+    /// Returns an asynchronous sequence that yields sliding windows of the given size
+    /// over the source sequence, each yielded as an array. The first window is emitted
+    /// once <c>windowSize</c> elements have been consumed; subsequent windows slide one
+    /// element at a time. The sequence is empty when the source has fewer than
+    /// <c>windowSize</c> elements. Raises <c>System.ArgumentException</c> if
+    /// <c>windowSize</c> is less than 1.
+    val windowed : windowSize:int -> source:AsyncSeq<'T> -> AsyncSeq<'T []>
+
     /// Asynchronously aggregate the elements of the input asynchronous sequence using the
     /// specified asynchronous 'aggregation' function.
     val foldAsync : folder:('State -> 'T -> Async<'State>) -> state:'State -> source:AsyncSeq<'T> -> Async<'State>
@@ -206,11 +214,50 @@ module AsyncSeq =
     /// specified 'aggregation' function.
     val fold : folder:('State -> 'T -> 'State) -> state:'State -> source:AsyncSeq<'T> -> Async<'State>
 
+    /// Asynchronously reduce the elements of the input asynchronous sequence using the
+    /// specified asynchronous 'reduction' function. Raises InvalidOperationException if the sequence is empty.
+    val reduceAsync : reduction:('T -> 'T -> Async<'T>) -> source:AsyncSeq<'T> -> Async<'T>
+
+    /// Asynchronously reduce the elements of the input asynchronous sequence using the
+    /// specified 'reduction' function. Raises InvalidOperationException if the sequence is empty.
+    val reduce : reduction:('T -> 'T -> 'T) -> source:AsyncSeq<'T> -> Async<'T>
+
     /// 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)
                                 and  ^T : (static member Zero : ^T)
 
+    /// Asynchronously sum the mapped elements of an asynchronous sequence using a synchronous projection.
+    val inline sumBy : projection:('T -> ^U) -> source:AsyncSeq<'T> -> Async< ^U>
+                                when ^U : (static member ( + ) : ^U * ^U -> ^U)
+                                and  ^U : (static member Zero : ^U)
+
+    /// Asynchronously sum the mapped elements of an asynchronous sequence using an asynchronous projection.
+    val inline sumByAsync : projection:('T -> Async< ^U>) -> source:AsyncSeq<'T> -> Async< ^U>
+                                when ^U : (static member ( + ) : ^U * ^U -> ^U)
+                                and  ^U : (static member Zero : ^U)
+
+    /// Asynchronously compute the average of the elements of the input asynchronous sequence.
+    /// Raises InvalidArgumentException if the sequence is empty.
+    val inline average : source:AsyncSeq< ^T> -> Async< ^T>
+                                when ^T : (static member ( + ) : ^T * ^T -> ^T)
+                                and  ^T : (static member DivideByInt : ^T * int -> ^T)
+                                and  ^T : (static member Zero : ^T)
+
+    /// Asynchronously compute the average of the mapped elements of an asynchronous sequence using a synchronous projection.
+    /// Raises InvalidArgumentException if the sequence is empty.
+    val inline averageBy : projection:('T -> ^U) -> source:AsyncSeq<'T> -> Async< ^U>
+                                when ^U : (static member ( + ) : ^U * ^U -> ^U)
+                                and  ^U : (static member DivideByInt : ^U * int -> ^U)
+                                and  ^U : (static member Zero : ^U)
+
+    /// Asynchronously compute the average of the mapped elements of an asynchronous sequence using an asynchronous projection.
+    /// Raises InvalidArgumentException if the sequence is empty.
+    val inline averageByAsync : projection:('T -> Async< ^U>) -> source:AsyncSeq<'T> -> Async< ^U>
+                                when ^U : (static member ( + ) : ^U * ^U -> ^U)
+                                and  ^U : (static member DivideByInt : ^U * int -> ^U)
+                                and  ^U : (static member Zero : ^U)
+
     /// Asynchronously determine if the sequence contains the given value
     val contains : value:'T -> source:AsyncSeq<'T> -> Async<bool> when 'T : equality
 

tests/FSharp.Control.AsyncSeq.Tests/AsyncSeqTests.fs

@@ -211,6 +211,52 @@ let ``AsyncSeq.sum works``() =
       let expected = ls |> List.sum
       Assert.True((expected = actual))
 
+[<Test>]
+let ``AsyncSeq.sumBy works``() =
+  for i in 0 .. 10 do
+      let ls = [ 1 .. i ]
+      let actual = AsyncSeq.ofSeq ls |> AsyncSeq.sumBy float |> Async.RunSynchronously
+      let expected = ls |> List.sumBy float
+      Assert.AreEqual(expected, actual)
+
+[<Test>]
+let ``AsyncSeq.sumByAsync works``() =
+  for i in 0 .. 10 do
+      let ls = [ 1 .. i ]
+      let actual = AsyncSeq.ofSeq ls |> AsyncSeq.sumByAsync (float >> async.Return) |> Async.RunSynchronously
+      let expected = ls |> List.sumBy float
+      Assert.AreEqual(expected, actual)
+
+[<Test>]
+let ``AsyncSeq.average works``() =
+  for i in 1 .. 10 do
+      let ls = [ 1.0 .. float i ]
+      let actual = AsyncSeq.ofSeq ls |> AsyncSeq.average |> Async.RunSynchronously
+      let expected = ls |> List.average
+      Assert.AreEqual(expected, actual)
+
+[<Test>]
+let ``AsyncSeq.average raises on empty sequence``() =
+  Assert.Throws<System.ArgumentException>(fun () ->
+      AsyncSeq.empty<float> |> AsyncSeq.average |> Async.RunSynchronously |> ignore
+  ) |> ignore
+
+[<Test>]
+let ``AsyncSeq.averageBy works``() =
+  for i in 1 .. 10 do
+      let ls = [ 1 .. i ]
+      let actual = AsyncSeq.ofSeq ls |> AsyncSeq.averageBy float |> Async.RunSynchronously
+      let expected = ls |> List.averageBy float
+      Assert.AreEqual(expected, actual)
+
+[<Test>]
+let ``AsyncSeq.averageByAsync works``() =
+  for i in 1 .. 10 do
+      let ls = [ 1 .. i ]
+      let actual = AsyncSeq.ofSeq ls |> AsyncSeq.averageByAsync (float >> async.Return) |> Async.RunSynchronously
+      let expected = ls |> List.averageBy float
+      Assert.AreEqual(expected, actual)
+
 
 [<Test>]
 let ``AsyncSeq.length works``() =
@@ -2329,6 +2375,45 @@ let ``AsyncSeq.fold with empty sequence should return seed``() =
                |> Async.RunSynchronously
   Assert.AreEqual(10, result)
 
+[<Test>]
+let ``AsyncSeq.reduce sums non-empty sequence`` () =
+  let result = asyncSeq { yield 1; yield 2; yield 3; yield 4; yield 5 }
+               |> AsyncSeq.reduce (+)
+               |> Async.RunSynchronously
+  Assert.AreEqual(15, result)
+
+[<Test>]
+let ``AsyncSeq.reduce single element returns that element`` () =
+  let result = asyncSeq { yield 42 }
+               |> AsyncSeq.reduce (+)
+               |> Async.RunSynchronously
+  Assert.AreEqual(42, result)
+
+[<Test>]
+let ``AsyncSeq.reduce empty sequence raises InvalidOperationException`` () =
+  Assert.Throws<InvalidOperationException>(fun () ->
+    AsyncSeq.empty<int>
+    |> AsyncSeq.reduce (+)
+    |> Async.RunSynchronously
+    |> ignore) |> ignore
+
+[<Test>]
+let ``AsyncSeq.reduceAsync accumulates with async function`` () =
+  async {
+    let result =
+      asyncSeq { yield 10; yield 3; yield 2 }
+      |> AsyncSeq.reduceAsync (fun a b -> async { return a - b })
+      |> Async.RunSynchronously
+    Assert.AreEqual(5, result)
+  } |> Async.RunSynchronously
+
+[<Test>]
+let ``AsyncSeq.reduce matches Seq.reduce`` () =
+  for ls in [ [1]; [1;2]; [3;1;4;1;5;9;2;6] ] do
+    let expected = Seq.reduce (+) ls
+    let actual = AsyncSeq.ofSeq ls |> AsyncSeq.reduce (+) |> Async.RunSynchronously
+    Assert.AreEqual(expected, actual)
+
 [<Test>]
 let ``AsyncSeq.ofSeq should work with large sequence``() =
   let largeSeq = seq { 1 .. 100 }
@@ -2553,6 +2638,46 @@ let ``AsyncSeq.pairwise with three elements should produce two pairs`` () =
   let result = AsyncSeq.pairwise source |> AsyncSeq.toListSynchronously
   Assert.AreEqual([(1, 2); (2, 3)], result)
 
+[<Test>]
+let ``AsyncSeq.windowed empty sequence returns empty`` () =
+  let result = AsyncSeq.windowed 3 AsyncSeq.empty<int> |> AsyncSeq.toListSynchronously
+  Assert.AreEqual([], result)
+
+[<Test>]
+let ``AsyncSeq.windowed fewer elements than window returns empty`` () =
+  let source = asyncSeq { yield 1; yield 2 }
+  let result = AsyncSeq.windowed 3 source |> AsyncSeq.toListSynchronously
+  Assert.AreEqual([], result)
+
+[<Test>]
+let ``AsyncSeq.windowed exact window size returns single window`` () =
+  let source = asyncSeq { yield 1; yield 2; yield 3 }
+  let result = AsyncSeq.windowed 3 source |> AsyncSeq.toListSynchronously
+  Assert.AreEqual([[|1; 2; 3|]], result)
+
+[<Test>]
+let ``AsyncSeq.windowed sliding window produces correct windows`` () =
+  let source = asyncSeq { yield 1; yield 2; yield 3; yield 4; yield 5 }
+  let result = AsyncSeq.windowed 3 source |> AsyncSeq.toListSynchronously
+  Assert.AreEqual([[|1;2;3|]; [|2;3;4|]; [|3;4;5|]], result)
+
+[<Test>]
+let ``AsyncSeq.windowed size 1 returns each element as singleton array`` () =
+  let source = asyncSeq { yield 10; yield 20; yield 30 }
+  let result = AsyncSeq.windowed 1 source |> AsyncSeq.toListSynchronously
+  Assert.AreEqual([[|10|]; [|20|]; [|30|]], result)
+
+[<Test>]
+let ``AsyncSeq.windowed size 2 is equivalent to pairwise as arrays`` () =
+  let source = asyncSeq { yield 1; yield 2; yield 3; yield 4 }
+  let result = AsyncSeq.windowed 2 source |> AsyncSeq.toListSynchronously
+  Assert.AreEqual([[|1;2|]; [|2;3|]; [|3;4|]], result)
+
+[<Test>]
+let ``AsyncSeq.windowed with size 0 raises ArgumentException`` () =
+  Assert.Throws<System.ArgumentException>(fun () ->
+    AsyncSeq.windowed 0 (asyncSeq { yield 1 }) |> AsyncSeq.toListSynchronously |> ignore) |> ignore
+
 [<Test>]
 let ``AsyncSeq.distinctUntilChangedWith should work with custom equality`` () =
   let source = asyncSeq { yield "a"; yield "A"; yield "B"; yield "b"; yield "c" }