Add distinctUntilChanged and tests

Author: njlr

src/FSharp.Control.TaskSeq.Test/FSharp.Control.TaskSeq.Test.fsproj

@@ -19,6 +19,7 @@
     <Compile Include="TaskSeq.Empty.Tests.fs" />
     <Compile Include="TaskSeq.ExactlyOne.Tests.fs" />
     <Compile Include="TaskSeq.Except.Tests.fs" />
+    <Compile Include="TaskSeq.DistinctUntilChanged.Tests.fs" />
     <Compile Include="TaskSeq.Exists.Tests.fs" />
     <Compile Include="TaskSeq.Filter.Tests.fs" />
     <Compile Include="TaskSeq.FindIndex.Tests.fs" />

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

@@ -0,0 +1,39 @@
+module TaskSeq.Tests.DistinctUntilChanged
+
+open Xunit
+open FsUnit.Xunit
+
+open FSharp.Control
+
+//
+// TaskSeq.distinctUntilChanged
+//
+
+
+module EmptySeq =
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChanged with null source raises`` () = assertNullArg <| fun () -> TaskSeq.distinctUntilChanged null
+
+    [<Theory; ClassData(typeof<TestEmptyVariants>)>]
+    let ``TaskSeq-distinctUntilChanged has no effect`` variant = task {
+        do!
+            Gen.getEmptyVariant variant
+            |> TaskSeq.distinctUntilChanged
+            |> TaskSeq.toListAsync
+            |> Task.map (List.isEmpty >> should be True)
+    }
+
+module Functionality =
+    [<Fact>]
+    let ``TaskSeq-distinctUntilChanged should return no consecutive duplicates`` () = task {
+        let ts =
+            [ 'A'; 'A'; 'B'; 'Z'; 'C'; 'C'; 'Z'; 'C'; 'D'; 'D'; 'D'; 'Z' ]
+            |> TaskSeq.ofList
+
+        let! xs = ts |> TaskSeq.distinctUntilChanged |> TaskSeq.toListAsync
+
+        xs
+        |> List.map string
+        |> String.concat ""
+        |> should equal "ABZCZCDZ"
+    }

src/FSharp.Control.TaskSeq/TaskSeq.fs

@@ -358,6 +358,8 @@ type TaskSeq private () =
     static member except itemsToExclude source = Internal.except itemsToExclude source
     static member exceptOfSeq itemsToExclude source = Internal.exceptOfSeq itemsToExclude source
 
+    static member distinctUntilChanged source = Internal.distinctUntilChanged source
+
     static member forall predicate source = Internal.forall (Predicate predicate) source
     static member forallAsync predicate source = Internal.forall (PredicateAsync predicate) source
 

src/FSharp.Control.TaskSeq/TaskSeq.fsi

@@ -1297,6 +1297,16 @@ type TaskSeq =
     /// <exception cref="T:ArgumentNullException">Thrown when either of the two input task sequences is null.</exception>
     static member exceptOfSeq<'T when 'T: equality> : itemsToExclude: seq<'T> -> source: TaskSeq<'T> -> TaskSeq<'T>
 
+    /// <summary>
+    /// Returns a new task sequence without consecutive duplicate elements.
+    /// </summary>
+    ///
+    /// <param name="source">The input task sequence whose consecutive duplicates will be removed.</param>
+    /// <returns>A sequence without consecutive duplicates elements.</returns>
+    ///
+    /// <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>
     /// Combines the two task sequences into a new task sequence of pairs. The two sequences need not have equal lengths:
     /// when one sequence is exhausted any remaining elements in the other sequence are ignored.

src/FSharp.Control.TaskSeq/TaskSeqInternal.fs

@@ -1097,3 +1097,22 @@ module internal TaskSeqInternal =
                     go <- step
 
         }
+
+    let distinctUntilChanged (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 previous = current then
+                        () // skip
+                    else
+                        yield current
+                        maybePrevious <- ValueSome current
+        }