+ DistinctUntilChanged

async-enumerable-dotnet-test/DistinctTest.cs

@@ -62,7 +62,7 @@ await AsyncEnumerable.Range(1, 5)
         public async void Custom_Set()
         {
             await AsyncEnumerable.Range(1, 5)
-                .Distinct(v => (v % 3), () => new HashSet<long>())
+                .Distinct(v => v % 3, () => new HashSet<long>())
                 .AssertResult(1, 2, 3);
         }
     }

async-enumerable-dotnet-test/DistinctUntilChangedTest.cs

@@ -0,0 +1,69 @@
+// Copyright (c) David Karnok & Contributors.
+// Licensed under the Apache 2.0 License.
+// See LICENSE file in the project root for full license information.
+
+using Xunit;
+using async_enumerable_dotnet;
+using System.Collections.Generic;
+
+namespace async_enumerable_dotnet_test
+{
+    public class DistinctUntilChangedTest
+    {
+        [Fact]
+        public async void Empty()
+        {
+            await AsyncEnumerable.Empty<int>()
+                .DistinctUntilChanged()
+                .AssertResult();
+        }
+
+        [Fact]
+        public async void Just()
+        {
+            await AsyncEnumerable.Just(1)
+                .DistinctUntilChanged()
+                .AssertResult(1);
+        }
+
+        [Fact]
+        public async void Range()
+        {
+            await AsyncEnumerable.Range(1, 5)
+                .DistinctUntilChanged()
+                .AssertResult(1, 2, 3, 4, 5);
+        }
+
+        [Fact]
+        public async void Redundant()
+        {
+            await AsyncEnumerable.FromArray(1, 2, 3, 3, 2, 1, 1, 4, 5, 4, 4)
+                .DistinctUntilChanged()
+                .AssertResult(1, 2, 3, 2, 1, 4, 5, 4);
+        }
+
+        [Fact]
+        public async void Redundant_Comparer()
+        {
+            await AsyncEnumerable.FromArray(1, 2, 3, 3, 2, 1, 1, 4, 5, 4, 4)
+                .DistinctUntilChanged(EqualityComparer<int>.Default)
+                .AssertResult(1, 2, 3, 2, 1, 4, 5, 4);
+        }
+
+        [Fact]
+        public async void KeySelector()
+        {
+            await AsyncEnumerable.Range(1, 10)
+                .DistinctUntilChanged(k => k / 3)
+                .AssertResult(1, 3, 6, 9);
+        }
+
+        [Fact]
+        public async void KeySelector_KeyComparer()
+        {
+            await AsyncEnumerable.Range(1, 10)
+                .DistinctUntilChanged(k => k / 3, EqualityComparer<long>.Default)
+                .AssertResult(1, 3, 6, 9);
+        }
+    }
+}

async-enumerable-dotnet/AsyncEnumerable.cs

@@ -1678,7 +1678,7 @@ public static IAsyncEnumerable<TSource> Distinct<TSource>(this IAsyncEnumerable<
 
         /// <summary>
         /// Checks if the source items have been seen before by checking
-        /// a key extracted from such items agains a set of keys and
+        /// a key extracted from such items against a set of keys and
         /// drops such items, ensuring that only distinct source items pass
         /// through.
         /// </summary>
@@ -1696,5 +1696,66 @@ public static IAsyncEnumerable<TSource> Distinct<TSource>(this IAsyncEnumerable<
             RequireNonNull(setSupplier, nameof(setSupplier));
             return new Distinct<TSource, TKey>(source, keySelector, setSupplier);
         }
+
+        /// <summary>
+        /// Checks if the current item is distinct from the previous item,
+        /// and if so it is no it is relayed.
+        /// </summary>
+        /// <typeparam name="TSource">The element type of the source and result sequences.</typeparam>
+        /// <param name="source">The source sequence to filter for distinct subsequent items.</param>
+        /// <returns>The new IAsyncEnumerable sequence.</returns>
+        public static IAsyncEnumerable<TSource> DistinctUntilChanged<TSource>(this IAsyncEnumerable<TSource> source)
+        {
+            return DistinctUntilChanged(source, v => v, EqualityComparer<TSource>.Default);
+        }
+
+        /// <summary>
+        /// Checks if the current item is distinct from the previous item,
+        /// and if so it is no it is relayed,
+        /// based on comparing the items via a custom equality comparer.
+        /// </summary>
+        /// <typeparam name="TSource">The element type of the source and result sequences.</typeparam>
+        /// <param name="source">The source sequence to filter for distinct subsequent items.</param>
+        /// <param name="comparer">The comparer for comparing the source items.</param>
+        /// <returns>The new IAsyncEnumerable sequence.</returns>
+        public static IAsyncEnumerable<TSource> DistinctUntilChanged<TSource>(this IAsyncEnumerable<TSource> source, IEqualityComparer<TSource> comparer)
+        {
+            return DistinctUntilChanged(source, v => v, comparer);
+        }
+
+        /// <summary>
+        /// Checks if the current item is distinct from the previous item,
+        /// and if so it is no it is relayed,
+        /// based on comparing keys extracted via a function.
+        /// </summary>
+        /// <typeparam name="TSource">The element type of the source and result sequences.</typeparam>
+        /// <typeparam name="TKey">The type of the keys extracted for comparison</typeparam>
+        /// <param name="source">The source sequence to filter for distinct subsequent items.</param>
+        /// <param name="keySelector">The function receiving the current source item and should return a key value for comparison.</param>
+        /// <returns>The new IAsyncEnumerable sequence.</returns>
+        public static IAsyncEnumerable<TSource> DistinctUntilChanged<TSource, TKey>(this IAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector)
+        {
+            return DistinctUntilChanged(source, keySelector, EqualityComparer<TKey>.Default);
+        }
+
+        /// <summary>
+        /// Checks if the current item is distinct from the previous item,
+        /// and if so it is no it is relayed,
+        /// based on comparing keys extracted via a function and using
+        /// a custom equality comparer.
+        /// </summary>
+        /// <typeparam name="TSource">The element type of the source and result sequences.</typeparam>
+        /// <typeparam name="TKey">The type of the keys extracted for comparison</typeparam>
+        /// <param name="source">The source sequence to filter for distinct subsequent items.</param>
+        /// <param name="keySelector">The function receiving the current source item and should return a key value for comparison.</param>
+        /// <param name="keyComparer">The comparer for comparing the key values extracted via <paramref name="keySelector"/>.</param>
+        /// <returns>The new IAsyncEnumerable sequence.</returns>
+        public static IAsyncEnumerable<TSource> DistinctUntilChanged<TSource, TKey>(this IAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector, IEqualityComparer<TKey> keyComparer)
+        {
+            RequireNonNull(source, nameof(source));
+            RequireNonNull(keySelector, nameof(keySelector));
+            RequireNonNull(keyComparer, nameof(keyComparer));
+            return new DistinctUntilChanged<TSource, TKey>(source, keySelector, keyComparer);
+        }
     }
 }

async-enumerable-dotnet/impl/DistinctUntilChanged.cs

@@ -0,0 +1,87 @@
+// Copyright (c) David Karnok & Contributors.
+// Licensed under the Apache 2.0 License.
+// See LICENSE file in the project root for full license information.
+
+using System;
+using System.Collections.Generic;
+using System.Threading.Tasks;
+
+namespace async_enumerable_dotnet.impl
+{
+    internal sealed class DistinctUntilChanged<TSource, TKey> : IAsyncEnumerable<TSource>
+    {
+        private readonly IAsyncEnumerable<TSource> _source;
+
+        private readonly Func<TSource, TKey> _keySelector;
+
+        private readonly IEqualityComparer<TKey> _keyComparer;
+
+        public DistinctUntilChanged(IAsyncEnumerable<TSource> source, Func<TSource, TKey> keySelector, IEqualityComparer<TKey> keyComparer)
+        {
+            _source = source;
+            _keySelector = keySelector;
+            _keyComparer = keyComparer;
+        }
+
+        public IAsyncEnumerator<TSource> GetAsyncEnumerator()
+        {
+            return new DistinctUntilChangedEnumerator(_source.GetAsyncEnumerator(), _keySelector, _keyComparer);
+        }
+
+        private sealed class DistinctUntilChangedEnumerator : IAsyncEnumerator<TSource>
+        {
+            private readonly IAsyncEnumerator<TSource> _source;
+
+            private readonly Func<TSource, TKey> _keySelector;
+
+            private readonly IEqualityComparer<TKey> _keyComparer;
+
+            private TKey _prevKey;
+
+            public TSource Current => _source.Current;
+
+            private bool _once;
+
+            public DistinctUntilChangedEnumerator(IAsyncEnumerator<TSource> source, Func<TSource, TKey> keySelector, IEqualityComparer<TKey> keyComparer)
+            {
+                _source = source;
+                _keySelector = keySelector;
+                _keyComparer = keyComparer;
+            }
+
+            public ValueTask DisposeAsync()
+            {
+                _prevKey = default;
+                return _source.DisposeAsync();
+            }
+
+            public async ValueTask<bool> MoveNextAsync()
+            {
+                for (; ;)
+                {
+                    if (await _source.MoveNextAsync())
+                    {
+                        var key = _keySelector(_source.Current);
+                        if (!_once)
+                        {
+                            _once = true;
+                            _prevKey = key;
+                            return true;
+                        }
+                        if (!_keyComparer.Equals(_prevKey, key))
+                        {
+                            _prevKey = key;
+                            return true;
+                        }
+
+                        _prevKey = key;
+                    }
+                    else
+                    {
+                        return false;
+                    }
+                }
+            }
+        }
+    }
+}