diff --git a/src/DynamicData.Tests/Cache/TransformAsyncFixture.cs b/src/DynamicData.Tests/Cache/TransformAsyncFixture.cs index a3890dcaa..6c258e6d9 100644 --- a/src/DynamicData.Tests/Cache/TransformAsyncFixture.cs +++ b/src/DynamicData.Tests/Cache/TransformAsyncFixture.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.ObjectModel; using System.Linq; using System.Reactive; using System.Reactive.Linq; @@ -88,6 +89,44 @@ public void Remove() stub.Results.Data.Count.Should().Be(0, "Should be nothing cached"); } + [Fact] + public async Task RemoveFlowsToTheEnd() + { + int transform = 0; + int count = 500; + ReadOnlyObservableCollection collection; + + var cache = new SourceCache(p => p.Name); + var people = Enumerable.Range(1, count).Select(l => new Person("Name" + l, l)).ToArray(); + + cache.Connect() + .TransformAsync(async person => + { + try + { + await Task.Delay(Random.Shared.Next(1, 12)); + return person; + } + finally + { + transform++; + } + }) + .Bind(out collection) + .Subscribe(); + + foreach (var p in people) + { + cache.AddOrUpdate(p); + cache.RemoveKey(p.Name); + } + + while (transform != count) + await Task.Delay(100); + await Task.Delay(3000); + collection.Count.Should().Be(0); + } + [Fact] public void ReTransformAll() { diff --git a/src/DynamicData/Cache/Internal/TransformAsync.cs b/src/DynamicData/Cache/Internal/TransformAsync.cs index de56282aa..626bbab39 100644 --- a/src/DynamicData/Cache/Internal/TransformAsync.cs +++ b/src/DynamicData/Cache/Internal/TransformAsync.cs @@ -2,11 +2,7 @@ // Roland Pheasant licenses this file to you under the MIT license. // See the LICENSE file in the project root for full license information. -using System; -using System.Linq; using System.Reactive.Linq; -using System.Threading; -using System.Threading.Tasks; using DynamicData.Kernel; @@ -38,36 +34,35 @@ public IObservable> Run() var cache = new ChangeAwareCache(); var asyncLock = new SemaphoreSlim(1, 1); - var transformer = _source.SelectMany(async changes => + var transformer = _source.Select(async changes => { try { await asyncLock.WaitAsync(); return await DoTransform(cache, changes).ConfigureAwait(false); - } finally { asyncLock.Release(); } - }); + }).Concat(); if (_forceTransform is not null) { var locker = new object(); - var forced = _forceTransform.Synchronize(locker).SelectMany(async shouldTransform => + var forced = _forceTransform.Synchronize(locker) + .Select(async shouldTransform => { try { await asyncLock.WaitAsync(); return await DoTransform(cache, shouldTransform).ConfigureAwait(false); - } finally { asyncLock.Release(); } - }); + }).Concat(); transformer = transformer.Synchronize(locker).Merge(forced); } @@ -159,9 +154,9 @@ public TransformedItemContainer(TSource source, TDestination destination) Destination = destination; } - public TSource Source { get; } - public TDestination Destination { get; } + + public TSource Source { get; } } private sealed class TransformResult