[Repo Assist] feat: add TaskSeq.distinctUntilChangedWith and distinctUntilChangedWithAsync

release-notes.txt

@@ -2,6 +2,7 @@
 Release notes:
 
 1.0.0
+    - adds TaskSeq.distinctUntilChangedWith and TaskSeq.distinctUntilChangedWithAsync, #345
     - adds TaskSeq.withCancellation, #167
     - adds docs/ with fsdocs-based documentation site covering generating, transforming, consuming, combining and advanced operations
 

src/FSharp.Control.TaskSeq.Test/TaskSeq.DistinctUntilChanged.Tests.fs

@@ -6,7 +6,7 @@ open FsUnit.Xunit
 open FSharp.Control
 
 //
-// TaskSeq.distinctUntilChanged
+// TaskSeq.distinctUntilChanged / distinctUntilChangedWith / distinctUntilChangedWithAsync
 //
 
 
@@ -23,6 +23,34 @@ module EmptySeq =
             |> Task.map (List.isEmpty >> should be True)
     }
 
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWith with null source raises`` () =
+        assertNullArg
+        <| fun () -> TaskSeq.distinctUntilChangedWith (fun _ _ -> false) null
+
+    [<Theory; ClassData(typeof<TestEmptyVariants>)>]
+    let ``TaskSeq-distinctUntilChangedWith has no effect on empty`` variant = task {
+        do!
+            Gen.getEmptyVariant variant
+            |> TaskSeq.distinctUntilChangedWith (fun _ _ -> false)
+            |> TaskSeq.toListAsync
+            |> Task.map (List.isEmpty >> should be True)
+    }
+
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWithAsync with null source raises`` () =
+        assertNullArg
+        <| fun () -> TaskSeq.distinctUntilChangedWithAsync (fun _ _ -> task { return false }) null
+
+    [<Theory; ClassData(typeof<TestEmptyVariants>)>]
+    let ``TaskSeq-distinctUntilChangedWithAsync has no effect on empty`` variant = task {
+        do!
+            Gen.getEmptyVariant variant
+            |> TaskSeq.distinctUntilChangedWithAsync (fun _ _ -> task { return false })
+            |> TaskSeq.toListAsync
+            |> Task.map (List.isEmpty >> should be True)
+    }
+
 module Functionality =
     [<Fact>]
     let ``TaskSeq-distinctUntilChanged should return no consecutive duplicates`` () = task {
@@ -90,6 +118,129 @@ module Functionality =
         xs |> should equal [ 1..10 ]
     }
 
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWith with structural equality comparer behaves like distinctUntilChanged`` () = task {
+        let ts =
+            [ 'A'; 'A'; 'B'; 'Z'; 'C'; 'C'; 'Z'; 'C'; 'D'; 'D'; 'D'; 'Z' ]
+            |> TaskSeq.ofList
+
+        let! xs =
+            ts
+            |> TaskSeq.distinctUntilChangedWith (=)
+            |> TaskSeq.toListAsync
+
+        xs
+        |> List.map string
+        |> String.concat ""
+        |> should equal "ABZCZCDZ"
+    }
+
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWith with always-true comparer returns only first element`` () = task {
+        let! xs =
+            taskSeq { yield! [ 1; 2; 3; 4; 5 ] }
+            |> TaskSeq.distinctUntilChangedWith (fun _ _ -> true)
+            |> TaskSeq.toListAsync
+
+        xs |> should equal [ 1 ]
+    }
+
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWith with always-false comparer returns all elements`` () = task {
+        let! xs =
+            taskSeq { yield! [ 1; 1; 2; 2; 3 ] }
+            |> TaskSeq.distinctUntilChangedWith (fun _ _ -> false)
+            |> TaskSeq.toListAsync
+
+        xs |> should equal [ 1; 1; 2; 2; 3 ]
+    }
+
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWith can use custom projection for equality`` () = task {
+        // Treat values as equal if their absolute difference is <= 1
+        let closeEnough a b = abs (a - b) <= 1
+
+        let! xs =
+            taskSeq { yield! [ 10; 11; 9; 20; 21; 5 ] }
+            |> TaskSeq.distinctUntilChangedWith closeEnough
+            |> TaskSeq.toListAsync
+
+        // 10≈11 skip; 11≈9 skip (|11-9|=2? no, |11-9|=2>1, so keep 9); 9 vs 20 keep; 20≈21 skip; 21 vs 5 keep
+        // Wait: |10-11|=1 skip 11; |10-9|=1 skip 9; 10 vs 20 keep 20; |20-21|=1 skip 21; 20 vs 5 keep 5
+        xs |> should equal [ 10; 20; 5 ]
+    }
+
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWith with single element returns singleton`` () = task {
+        let! xs =
+            taskSeq { yield 99 }
+            |> TaskSeq.distinctUntilChangedWith (fun _ _ -> true)
+            |> TaskSeq.toListAsync
+
+        xs |> should equal [ 99 ]
+    }
+
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWith case-insensitive string comparison`` () = task {
+        let! xs =
+            taskSeq { yield! [ "Hello"; "hello"; "HELLO"; "World"; "world" ] }
+            |> TaskSeq.distinctUntilChangedWith (fun a b -> System.String.Compare(a, b, System.StringComparison.OrdinalIgnoreCase) = 0)
+            |> TaskSeq.toListAsync
+
+        xs |> should equal [ "Hello"; "World" ]
+    }
+
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWithAsync with structural equality behaves like distinctUntilChanged`` () = task {
+        let ts = [ 1; 1; 2; 3; 3; 4 ] |> TaskSeq.ofList
+
+        let! xs =
+            ts
+            |> TaskSeq.distinctUntilChangedWithAsync (fun a b -> task { return a = b })
+            |> TaskSeq.toListAsync
+
+        xs |> should equal [ 1; 2; 3; 4 ]
+    }
+
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWithAsync with always-true async comparer returns only first element`` () = task {
+        let! xs =
+            taskSeq { yield! [ 10; 20; 30 ] }
+            |> TaskSeq.distinctUntilChangedWithAsync (fun _ _ -> task { return true })
+            |> TaskSeq.toListAsync
+
+        xs |> should equal [ 10 ]
+    }
+
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWithAsync with always-false async comparer returns all elements`` () = task {
+        let! xs =
+            taskSeq { yield! [ 5; 5; 5 ] }
+            |> TaskSeq.distinctUntilChangedWithAsync (fun _ _ -> task { return false })
+            |> TaskSeq.toListAsync
+
+        xs |> should equal [ 5; 5; 5 ]
+    }
+
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWithAsync can perform async work in comparer`` () = task {
+        let mutable comparerCallCount = 0
+
+        let asyncComparer a b = task {
+            comparerCallCount <- comparerCallCount + 1
+            return a = b
+        }
+
+        let! xs =
+            taskSeq { yield! [ 1; 1; 2; 2; 3 ] }
+            |> TaskSeq.distinctUntilChangedWithAsync asyncComparer
+            |> TaskSeq.toListAsync
+
+        xs |> should equal [ 1; 2; 3 ]
+        // comparer called for each pair of consecutive elements (4 pairs for 5 elements)
+        comparerCallCount |> should equal 4
+    }
+
 module SideEffects =
     [<Fact>]
     let ``TaskSeq-distinctUntilChanged consumes every element exactly once`` () = task {
@@ -132,3 +283,41 @@ module SideEffects =
 
         xs |> should equal [ 1..10 ]
     }
+
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWith consumes every element exactly once`` () = task {
+        let mutable count = 0
+
+        let ts = taskSeq {
+            for i in 1..5 do
+                count <- count + 1
+                yield i
+        }
+
+        let! xs =
+            ts
+            |> TaskSeq.distinctUntilChangedWith (fun a b -> a = b)
+            |> TaskSeq.toListAsync
+
+        count |> should equal 5
+        xs |> should equal [ 1; 2; 3; 4; 5 ]
+    }
+
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChangedWithAsync consumes every element exactly once`` () = task {
+        let mutable count = 0
+
+        let ts = taskSeq {
+            for i in 1..5 do
+                count <- count + 1
+                yield i
+        }
+
+        let! xs =
+            ts
+            |> TaskSeq.distinctUntilChangedWithAsync (fun a b -> task { return a = b })
+            |> TaskSeq.toListAsync
+
+        count |> should equal 5
+        xs |> should equal [ 1; 2; 3; 4; 5 ]
+    }

src/FSharp.Control.TaskSeq/TaskSeq.fs

@@ -476,6 +476,8 @@ type TaskSeq private () =
     static member distinctByAsync projection source = Internal.distinctByAsync projection source
 
     static member distinctUntilChanged source = Internal.distinctUntilChanged source
+    static member distinctUntilChangedWith comparer source = Internal.distinctUntilChangedWith comparer source
+    static member distinctUntilChangedWithAsync comparer source = Internal.distinctUntilChangedWithAsync comparer source
     static member pairwise source = Internal.pairwise source
     static member chunkBySize chunkSize source = Internal.chunkBySize chunkSize source
     static member windowed windowSize source = Internal.windowed windowSize source

src/FSharp.Control.TaskSeq/TaskSeq.fsi

@@ -1503,6 +1503,33 @@ type TaskSeq =
     /// <exception cref="T:ArgumentNullException">Thrown when the input task sequences is null.</exception>
     static member distinctUntilChanged<'T when 'T: equality> : source: TaskSeq<'T> -> TaskSeq<'T>
 
+    /// <summary>
+    /// Returns a new task sequence without consecutive duplicate elements, using the supplied <paramref name="comparer" />
+    /// to determine equality of consecutive elements. The comparer returns <see langword="true" /> if two elements are
+    /// considered equal (and thus the second should be skipped).
+    /// </summary>
+    ///
+    /// <param name="comparer">A function that returns <see langword="true" /> if two consecutive elements are equal.</param>
+    /// <param name="source">The input task sequence whose consecutive duplicates will be removed.</param>
+    /// <returns>A sequence without consecutive duplicate elements.</returns>
+    ///
+    /// <exception cref="T:ArgumentNullException">Thrown when the input task sequence is null.</exception>
+    static member distinctUntilChangedWith: comparer: ('T -> 'T -> bool) -> source: TaskSeq<'T> -> TaskSeq<'T>
+
+    /// <summary>
+    /// Returns a new task sequence without consecutive duplicate elements, using the supplied async <paramref name="comparer" />
+    /// to determine equality of consecutive elements. The comparer returns <see langword="true" /> if two elements are
+    /// considered equal (and thus the second should be skipped).
+    /// </summary>
+    ///
+    /// <param name="comparer">An async function that returns <see langword="true" /> if two consecutive elements are equal.</param>
+    /// <param name="source">The input task sequence whose consecutive duplicates will be removed.</param>
+    /// <returns>A sequence without consecutive duplicate elements.</returns>
+    ///
+    /// <exception cref="T:ArgumentNullException">Thrown when the input task sequence is null.</exception>
+    static member distinctUntilChangedWithAsync:
+        comparer: ('T -> 'T -> #Task<bool>) -> source: TaskSeq<'T> -> TaskSeq<'T>
+
     /// <summary>
     /// Returns a task sequence of each element in the source paired with its successor.
     /// The sequence is empty if the source has fewer than two elements.

src/FSharp.Control.TaskSeq/TaskSeqInternal.fs

@@ -1367,6 +1367,46 @@ module internal TaskSeqInternal =
                         maybePrevious <- ValueSome current
         }
 
+    let distinctUntilChangedWith (comparer: 'T -> 'T -> bool) (source: TaskSeq<_>) =
+        checkNonNull (nameof source) source
+
+        taskSeq {
+            let mutable maybePrevious = ValueNone
+
+            for current in source do
+                match maybePrevious with
+                | ValueNone ->
+                    yield current
+                    maybePrevious <- ValueSome current
+                | ValueSome previous ->
+                    if comparer previous current then
+                        () // skip
+                    else
+                        yield current
+                        maybePrevious <- ValueSome current
+        }
+
+    let distinctUntilChangedWithAsync (comparer: 'T -> 'T -> #Task<bool>) (source: TaskSeq<_>) =
+        checkNonNull (nameof source) source
+
+        taskSeq {
+            let mutable maybePrevious = ValueNone
+
+            for current in source do
+                match maybePrevious with
+                | ValueNone ->
+                    yield current
+                    maybePrevious <- ValueSome current
+                | ValueSome previous ->
+                    let! areEqual = comparer previous current
+
+                    if areEqual then
+                        () // skip
+                    else
+                        yield current
+                        maybePrevious <- ValueSome current
+        }
+
     let pairwise (source: TaskSeq<_>) =
         checkNonNull (nameof source) source