Feature: Add FilterOnObservable for Cache ChangeSets

[dwcullop]

Description

This PR implements the FilterOnObservable operator for Cache ChangeSets (the operator already exists exists for List ChangeSets). It filters a ChangeSet by using the given factory function to create an IObservable<bool> for each item that controls whether or not that item passes the filter. No items will pass the filter until the associated observable fires true.

This operator enables filtering changesets by something more complicated than a simple property value. For example, if the items in the ChangeSet have a sub-cache, you can use it to filter them by the contents of their individual sub-caches.

I originally had cloned the code from AutoRefresh, but I changed to just using the AutoRefreshOnObservable operator to reduce the code size because it is exactly what is needed.

Examples

Filter by main cache by the number of items in the sub-cache:

interface IPerson
{
    int Id { get; }
    string Name { get; }
    IObservableCache<IPerson, int> Children { get; }
}

var peopleDirectory = new SourceCache<IPerson, int>(p => p.Id);

// Observe the source cache for people with at least 2 children
using var cleanup = peopleDirectory.Connect()
                                   .FilterOnObservable(p => p.Children.CountChanged.Select(count => count >= 2))
                                   .OnItemAdded(p => Console.WriteLine($"{p.Name} has at least 2 kids!"))
                                   .Subscribe();

Create a view of the SourceCache based on how long ago the items were added to the cache (useful if you don't want stale items but don't want to remove them for the main cache).

IObservable<bool> CreateTimerFilter(TimeSpan timeSpan, bool filterResult) =>
    Observable.Timer(timeSpan).Select(_ => filterResult).StartWith(!filterResult);

// Special handling of entries older than 5 minutes
using var cleanup1 = observableCache.Connect()
                                    .FilterOnObservable(_ => CreateTimerFilter(TimeSpan.FromMinutes(5), true))
                                    .OnItemAdded(entry => HandleStaleEntry(entry))
                                    .Subscribe();

// Show brand new entries in a special UI list
using var cleanup2 = observableCache.Connect()
                                    .FilterOnObservable(_ => CreateTimerFilter(TimeSpan.FromSeconds(30), false))
                                    .ObserveOnDispatcher()
                                    .Bind(var out newEntryList)
                                    .Subscribe();

Unit Tests

I've added unit tests but please let me know if there are any use-cases that I've missed.

[codecov]

Codecov Report

Merging #735 (7cf9b33) into main (0be0d95) will increase coverage by 0.05%.
The diff coverage is 100.00%.

@@            Coverage Diff             @@
##             main     #735      +/-   ##
==========================================
+ Coverage   72.43%   72.48%   +0.05%     
==========================================
  Files         220      221       +1     
  Lines       11154    11176      +22     
==========================================
+ Hits         8079     8101      +22     
  Misses       3075     3075              

Files	Coverage Δ
...c/DynamicData/Cache/Internal/FilterOnObservable.cs	100.00% <100.00%> (ø)
src/DynamicData/Cache/ObservableCacheEx.cs	52.43% <100.00%> (+0.19%)	⬆️

[dwcullop]

This implements one of the requests from #409, but that issue has been closed by @RolandPheasant because the list version has issues. I'd like to understand the nature of the issues to make sure this PR doesn't have the same problems. If it doesn't, this implementation might be easily ported to support Lists.

[RolandPheasant]

This implements one of the requests from #409, but that issue has been closed by @RolandPheasant because the list version has issues. I'd like to understand the nature of the issues to make sure this PR doesn't have the same problems. If it doesn't, this implementation might be easily ported to support Lists.

If my memory serves my right, the issues were mainly down to performance and scalability. I think the main reason is due to the difficulty with optimising observables of unknow nature. However, my thought is if it exists with the list side of the system then it should also exist here.

Also maybe we should create some additional wiki pages expressing scalability concerns of certain operators. Although I suspect in most cases people are not using 50k records in a dataset and scalability is not everyone's concern.

[dwcullop]

If my memory serves my right, the issues were mainly down to performance and scalability. I think the main reason is due to the difficulty with optimising observables of unknow nature. However, my thought is if it exists with the list side of the system then it should also exist here.

Cache is always going to give better performance, so it might not be an issue. This implementation uses AutoRefreshOnObservable. If Unknown Observables are okay for that operator, then they should be okay here. I can add some performance testing to see if there is an issue.

Also maybe we should create some additional wiki pages expressing scalability concerns of certain operators. Although I suspect in most cases people are not using 50k records in a dataset and scalability is not everyone's concern.

I agree with you 100% on this. Just because some people might use it with heavy operators and end up with scalability issues when they have a large dataset, it doesn't mean that it won't be useful to some people who use simple observables and/or smaller datasets. All of these things are situational. The solution is to document the ones that might cause scalability issues and let the consumers decide what will work best for them.

Such things might be mitigated by using a hot observable that was going to run anyway. Then the overhead from filtering is minimal.

I'll do some perf testing and see if there's a problem and try to optimize the implementation as much as I can. I can't promise the approach taken will work for Lists, but Caches always have the advantage with regard to performance. Even if you have to deprecate the operator for lists, I would still like to see it for Cache because I use it A LOT in my code base (and having a native operator would be a huge improvement).

[RolandPheasant]

Looks good to me. Is this your last PR for now?

If so, would you mind bumping the version to 8.1 so we can release. See version.json#L2.

[dwcullop]

I'm done for now. I was working on one more, but I'm not sure when I'll have time to finish it (still needs unit tests). I got distracted by trying to optimize this PR (and failing miserably).

To improve this PR, I thought I had a way to eliminate the two Transforms but wasn't able to make it work (yet). Instead of the Filter Proxy object, I tried putting the bools for each object into a shared dictionary. However, that added a nasty race condition I haven't been able to nail down yet.

I'll rev the version in this PR and save those for another day.

---

The other PR adds operators for IObservable<Optional<T>> that are available for Optional<T> ... Convert, ValueOr, SelectValues, etc. so you can write code for Optional and Observable Optional that's essentially the same. I use them in my own projects, so I figured you and/or others might appreciate them as well. Much easier to read without all the extra Select and Filter calls in your Rx statements.

IObservable<string>  FileNameObservable();
Optional<FileInfo> LoadFile(string file);
Optional<Image> LoadImage(FileInfo fi);

// This
IObservable<Image> imageObservable = FileNameObservable().Convert(LoadFile).Convert(LoadImage).SelectValues();
     
// Instead of This
IObservable<Image> imageObservable = FileNameObservable().Select(opt => opt.Convert(LoadFile)).Select(opt => opt.Convert(LoadImage))
                                                         .Where(opt => opt.HasValue).Select(opt => opt.Value);

Also, if there are any open issues that you think are more urgent than others, feel free to tag me and I'll see if I can help out with them. I was thinking about trying to address the TransformManyAsync request (#461) by cloning TransformMany and then adapting it to be async. I am also considering something like:

IObservable<IChangeSet<TObject, TKey>> MergeChangeSets<TObject, TKey>(this IEnumerable<IObservable<IChangeSet<TObject, TKey>>> observableChangeSets);

[RolandPheasant]

Yet another excellent addition to the library. Thanks

[RolandPheasant]

@dwcullop all your changes have been deployed. Thanks for an excellent contribution to Dynamic Data.

Are you on the slack channel? It's very useful for discussions. See https://reactiveui.net/slack, and look for the #dynamicdata sub channel.

[dwcullop]

@dwcullop all your changes have been deployed. Thanks for an excellent contribution to Dynamic Data.

Thanks. I have a few more ideas and will contribute more as I am able to find the time.

Are you on the slack channel? It's very useful for discussions. See https://reactiveui.net/slack, and look for the #dynamicdata sub channel.

Never used Slack before but I'll check it out.

[github-actions]

This pull request has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.