-
-
Notifications
You must be signed in to change notification settings - Fork 182
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Feature: TransformOnObservable Operator for SourceCache (#841)
* TransformOnObservable operators - Transforms each item in a changeset to an `IObservable<TOther>` and that observable is used to supply the latest value for the corresponding key in the resulting transformed changeset * Improvements to OnItem* operators - Overloads that take an Action that uses the Key - Now use common implementation - OnItemRemoved doesn't use special operator unless necessary
- Loading branch information
Showing
5 changed files
with
404 additions
and
34 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
164 changes: 164 additions & 0 deletions
164
src/DynamicData.Tests/Cache/TransformOnObservableFixture.cs
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,164 @@ | ||
| using System; | ||
| using System.Linq; | ||
| using System.Reactive; | ||
| using System.Reactive.Linq; | ||
| using System.Threading.Tasks; | ||
| using Bogus; | ||
| using DynamicData.Kernel; | ||
| using DynamicData.Tests.Domain; | ||
| using DynamicData.Tests.Utilities; | ||
|
|
||
| using FluentAssertions; | ||
|
|
||
| using Xunit; | ||
|
|
||
| namespace DynamicData.Tests.Cache; | ||
|
|
||
| public class TransformOnObservableFixture : IDisposable | ||
| { | ||
| #if DEBUG | ||
| private const int InitialCount = 7; | ||
| private const int AddCount = 5; | ||
| private const int RemoveCount = 3; | ||
| private const int UpdateCount = 2; | ||
| #else | ||
| private const int InitialCount = 103; | ||
| private const int AddCount = 53; | ||
| private const int RemoveCount = 37; | ||
| private const int UpdateCount = 31; | ||
| #endif | ||
| private static readonly TimeSpan UpdateTime = TimeSpan.FromMilliseconds(50); | ||
|
|
||
| private readonly ISourceCache<Animal, int> _animalCache = new SourceCache<Animal, int>(a => a.Id); | ||
| private readonly ChangeSetAggregator<Animal, int> _animalResults; | ||
| private readonly Faker<Animal> _animalFaker; | ||
| private readonly Randomizer _randomizer = new (0x2112_2112); | ||
|
|
||
| public TransformOnObservableFixture() | ||
| { | ||
| _animalFaker = Fakers.Animal.Clone().WithSeed(_randomizer); | ||
| _animalCache.AddOrUpdate(_animalFaker.Generate(InitialCount)); | ||
| _animalResults = _animalCache.Connect().AsAggregator(); | ||
| } | ||
|
|
||
| [Fact] | ||
| public void ResultContainsAllInitialChildren() | ||
| { | ||
| // Arrange | ||
|
|
||
| // Act | ||
| using var results = _animalCache.Connect().TransformOnObservable((ani, id) => Observable.Return(ani.Name)).AsAggregator(); | ||
|
|
||
| // Assert | ||
| _animalResults.Data.Count.Should().Be(InitialCount); | ||
| results.Data.Count.Should().Be(InitialCount); | ||
| results.Messages.Count.Should().Be(1, "The child observables fire on subscription so everything should appear as a single changeset"); | ||
| } | ||
|
|
||
| [Fact] | ||
| public void ResultContainsAddedValues() | ||
| { | ||
| // Arrange | ||
| using var results = _animalCache.Connect().TransformOnObservable((ani, id) => Observable.Return(ani.Name)).AsAggregator(); | ||
|
|
||
| // Act | ||
| _animalCache.AddOrUpdate(_animalFaker.Generate(AddCount)); | ||
|
|
||
| // Assert | ||
| _animalResults.Data.Count.Should().Be(InitialCount + AddCount); | ||
| results.Data.Count.Should().Be(_animalResults.Data.Count); | ||
| results.Messages.Count.Should().Be(2, "Initial Adds and then the subsequent Additions should each be a single message"); | ||
| } | ||
|
|
||
| [Fact] | ||
| public void ResultDoesNotContainRemovedValues() | ||
| { | ||
| // Arrange | ||
| using var results = _animalCache.Connect().TransformOnObservable((ani, id) => Observable.Return(ani.Name)).AsAggregator(); | ||
|
|
||
| // Act | ||
| _animalCache.RemoveKeys(_randomizer.ListItems(_animalCache.Items.ToList(), RemoveCount).Select(a => a.Id)); | ||
|
|
||
| // Assert | ||
| _animalResults.Data.Count.Should().Be(InitialCount - RemoveCount); | ||
| results.Data.Count.Should().Be(_animalResults.Data.Count); | ||
| results.Messages.Count.Should().Be(2, "1 for Adds and 1 for Removes"); | ||
| } | ||
|
|
||
| [Fact] | ||
| public async Task ResultUpdatesOnFutureValues() | ||
| { | ||
| // Create an observable that fires a wrong value on an interval a fixed number of times | ||
| // then fires the expected value before completing | ||
| IObservable<string> CreateChildObs(Animal a, int id) => | ||
| Observable.Interval(UpdateTime) | ||
| .Select(n => $"{a.Name}-{id}-{n}") | ||
| .Take(UpdateCount) | ||
| .Concat(Observable.Return(a.Name)); | ||
|
|
||
| // Arrange | ||
| var shared = _animalCache.Connect().TransformOnObservable(CreateChildObs).Publish(); | ||
| using var results = shared.AsAggregator(); | ||
| var task = Task.Run(async () => await shared); | ||
| using var cleanup = shared.Connect(); | ||
| _animalCache.Dispose(); | ||
|
|
||
| // Act | ||
| await task; | ||
|
|
||
| // Assert | ||
| _animalResults.Data.Count.Should().Be(InitialCount); | ||
| results.Data.Count.Should().Be(_animalResults.Data.Count); | ||
| results.Summary.Overall.Adds.Should().Be(InitialCount); | ||
| results.Summary.Overall.Updates.Should().Be(InitialCount * UpdateCount, $"Each item should update {UpdateCount} times"); | ||
| results.Messages.Count.Should().BeGreaterThanOrEqualTo(1, "The delay may cause the messages to appear as multiple changesets"); | ||
| _animalCache.Items.ForEach(animal => results.Data.Lookup(animal.Id).Should().Be(Optional.Some(animal.Name))); | ||
| } | ||
|
|
||
| [Theory] | ||
| [InlineData(false, false)] | ||
| [InlineData(false, true)] | ||
| [InlineData(true, false)] | ||
| [InlineData(true, true)] | ||
| public void ResultCompletesOnlyWhenSourceAndAllChildrenComplete(bool completeSource, bool completeChildren) | ||
| { | ||
| IObservable<string> CreateChildObs(Animal a, int id) => | ||
| completeChildren | ||
| ? Observable.Return(a.Name) | ||
| : Observable.Return(a.Name).Concat(Observable.Never<string>()); | ||
|
|
||
| // Arrange | ||
| using var results = _animalCache.Connect().TransformOnObservable(CreateChildObs).AsAggregator(); | ||
|
|
||
| // Act | ||
| if (completeSource) | ||
| { | ||
| _animalCache.Dispose(); | ||
| } | ||
|
|
||
| // Assert | ||
| _animalResults.IsCompleted.Should().Be(completeSource); | ||
| results.IsCompleted.Should().Be(completeSource && completeChildren); | ||
| } | ||
|
|
||
| [Fact] | ||
| public void ResultFailsIfSourceFails() | ||
| { | ||
| // Arrange | ||
| var expectedError = new Exception("Expected"); | ||
| var throwObservable = Observable.Throw<IChangeSet<Animal, int>>(expectedError); | ||
| using var results = _animalCache.Connect().Concat(throwObservable).TransformOnObservable(animal => Observable.Return(animal)).AsAggregator(); | ||
|
|
||
| // Act | ||
| _animalCache.Dispose(); | ||
|
|
||
| // Assert | ||
| results.Error.Should().Be(expectedError); | ||
| } | ||
|
|
||
| public void Dispose() | ||
| { | ||
| _animalCache.Dispose(); | ||
| _animalResults.Dispose(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| // Copyright (c) 2011-2023 Roland Pheasant. All rights reserved. | ||
| // 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.Reactive.Disposables; | ||
| using System.Reactive.Linq; | ||
| using DynamicData.Internal; | ||
| using DynamicData.Kernel; | ||
|
|
||
| namespace DynamicData.Cache.Internal; | ||
|
|
||
| internal sealed class TransformOnObservable<TSource, TKey, TDestination>(IObservable<IChangeSet<TSource, TKey>> source, Func<TSource, TKey, IObservable<TDestination>> transform) | ||
| where TSource : notnull | ||
| where TKey : notnull | ||
| where TDestination : notnull | ||
| { | ||
| public IObservable<IChangeSet<TDestination, TKey>> Run() => Observable.Create<IChangeSet<TDestination, TKey>>(observer => | ||
| { | ||
| var cache = new ChangeAwareCache<TDestination, TKey>(); | ||
| var locker = new object(); | ||
| var pendingUpdates = 0; | ||
|
|
||
| // Helper to emit any pending changes when all the updates have been handled | ||
| void EmitChanges() | ||
| { | ||
| if (Interlocked.Decrement(ref pendingUpdates) == 0) | ||
| { | ||
| var changes = cache!.CaptureChanges(); | ||
| if (changes.Count > 0) | ||
| { | ||
| observer.OnNext(changes); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // Create the sub-observable that takes the result of the transformation, | ||
| // filters out unchanged values, and then updates the cache | ||
| IObservable<TDestination> CreateSubObservable(TSource obj, TKey key) => | ||
| transform(obj, key) | ||
| .DistinctUntilChanged() | ||
| .Do(_ => Interlocked.Increment(ref pendingUpdates)) | ||
| .Synchronize(locker!) | ||
| .Do(val => cache!.AddOrUpdate(val, key)); | ||
|
|
||
| // Always increment the counter OUTSIDE of the lock to signal any thread currently holding the lock | ||
| // to not emit the changeset because more changes are incoming. | ||
| var shared = source | ||
| .Do(_ => Interlocked.Increment(ref pendingUpdates)) | ||
| .Synchronize(locker!) | ||
| .Publish(); | ||
|
|
||
| // Use MergeMany because it automatically handles Add/Update/Remove and OnCompleted/OnError correctly | ||
| var subMerged = shared | ||
| .MergeMany(CreateSubObservable) | ||
| .SubscribeSafe(_ => EmitChanges(), observer.OnError, observer.OnCompleted); | ||
|
|
||
| // Subscribe to the shared Observable to handle Remove events. MergeMany will unsubscribe from the sub-observable, | ||
| // but the corresponding key value needs to be removed from the Cache so the remove is observed downstream. | ||
| var subRemove = shared | ||
| .OnItemRemoved((_, key) => cache!.Remove(key), invokeOnUnsubscribe: false) | ||
| .SubscribeSafe(_ => EmitChanges()); | ||
|
|
||
| return new CompositeDisposable(shared.Connect(), subMerged, subRemove); | ||
| }); | ||
| } |
Oops, something went wrong.