How to make list and cache disposal more fluent?

[glen-nicol]

I feel like I am missing something. I've found DisposeMany which works on the inner items of a change set stream. But I'm really struggling with intermediate lists/caches in a complex stream without ugly Observable.Using calls to include some cleanup. Is there a better way?

For example, I have a service that produces new immutable states that I enumerate into a list of items with a key. I want the new list to replace the old list of items so that items that are removed get removed instead of remaining in a stale state.

So I have:

service.StateOverTime()
.Select(t =>
    CreateVms(t) // IEnumerable
    .ToObservable()
    .ToObservableChangeSet()
    .AsObservableList()) // 👈 how to dispose of this?
.Switch() // replace old list 
.AddKey(x => x.Key)
.AsObservableCache();

My understanding of the SourceList<T> used in AsObservableList() is that it doesn't self dispose when the source IObservable<IChangeSet<T>> terminates.

Questions:

1. Do lists and caches need to be disposed when they are at the end of a stream?
   If nothing is watching what they are producing does disposal do anything worthwhile?
2. I imagine a leak will occur when they have downstream observers that expect a Completed/error message.
3. Is there an existing operator that can fluently cleanup lists and caches when a connection to it is unsubscribed?

[JakenVeina]

Do lists and caches need to be disposed when they are at the end of a stream? If nothing is watching what they are producing does disposal do anything worthwhile?

The general wisdom for RX in general, in .NET, is that you don't really need to dispose of source objects, like Subject<> (and the reason SourceList<> and SourceCache<> in DynamicData are IDisposable is because they keep Subject<>s internally). But it's still usually considered good practice to consider yourself responsible for disposing of IDisposables that you create. When it comes to a Subject<>, what .Dispose() actually does is force the subject to clear out its internal list of IObserver<>s, and then ensure the subject can't be used any further, by throwing, if any impure methods are called. Realistically, this doesn't really accomplish anything, because the whole Subject is (in theory) about to be put up for GC anyway, so all those references would get collected as well. There are those who think the use of IDisposable within System.Reactive was a mistake, or at least that it's over-used, for things that don't really represent resources that need to be disposed.

Bottom line, it's worth disposing of your source objects, for clarity and cleanliness, but it doesn't actually really matter, if all other best-practices are being followed.

I imagine a leak will occur when they have downstream observers that expect a Completed/error message.

A leak would only occur if you have a source object that you don't intend to ever use again, but don't dispose it AND don't allow it to be GC'd AND the only thing holding the downstream object alive is the reference that the source has. Subject<>s don't have any ACTUAL disposable resources, they just have object references that will become eligible for GC when the whole object is GC'd.

Is there an existing operator that can fluently cleanup lists and caches when a connection to it is unsubscribed?

In your example, it's as simple as adding a call to DisposeMany().

service.StateOverTime()
    .Select(t => CreateVms(t)
        .ToObservable()
        .ToObservableChangeSet()
        .AsObservableList()
        .DisposeMany())
    .Switch()
    .AddKey(x => x.Key)
    .AsObservableCache();

My understanding of the SourceList used in AsObservableList() is that it doesn't self dispose when the source IObservable<IChangeSet> terminates.

.AsObservableList() does indeed use a SourceList<> under the hood, but it's not exposed to you that way, because you don't need to worry about it. The operator in-fact does self-dispose the SourceList<> when the stream is torn down, and .DisposeMany() similarly disposes all items that haven't yet been removed, upon teardown.

---

On an entirely separate note, I'd be curious to know what your usage scenario here is, because constantly rebuilding these collections of VMs seems horribly-inefficient. Pretty much anything involving the "List" side of Dynamic Data is inefficient, and not recommended. You should really avoid using Dynamic Data Lists as much as possible, unless it's truly impossible for you to derive keys for a set of items.

[glen-nicol]

Direct responses then explanation of my actual goal is at the end.

I think the problem is my true source is a service that is long running. And commonly in my code the list/cache is an intermediate step based on a subscription to that service. So if I don't dispose of them, they never unsubscribe and then leak. Granted in the example I gave, the list is attached to an IEnumerable<> so if what you say is true, then this one shouldn't need cleanup.

For the record, .DisposeMany() cannot be used after AsObservableList(). At least in v 8.3.27. DisposeMany() expects IObservable<IChangeSet<T>> And calling Connect() to get one of those makes the outer call to Switch() use a different method that I am not sure behaves the same. I haven't tested that yet.

What I actually want:
I'm trying to group items with a header that has a some label text based on some data common to each group of items. On each data update groups and items in each group could be added, updated, removed.

Since the items are the real meat, my first attempt (not what was shown above) focused on the item viewmodels. I attempted to use a keyed cache to get each item viewmodel to replace equivalent ones as they get updated and then I used RemoveKey to flatten the items and used GroupOn to then divy out the item viewmodels into groups. This worked in many cases except when a group header was changed or a group was removed. The stale group would stick around. I used the header viewmodel as a key. I realize this is the source of the bug because the header text was contributing to the group key. But I couldn't find a clean way to access the first element of a group synchronously unless it was part of the key. This is needed because the header text is used to sort the groups and switching to an async property based on the group.List caused the sorter threw NRE because it executed before the groups list produced an element)

So I ended up just removing the second layer of ObservableCollection bindings in each group viewmodel and instead just generating the group with the new items each time. I agree its not as efficient on the UI side but in this case there is no state in the viewmodels. They are basically immutable display only fields.

[JakenVeina]

And calling Connect() to get one of those makes the outer call to Switch() use a different method that I am not sure behaves the same.

Aha! Okay, so I didn't realize there's a dedicated .Switch() operator in DD for switching IObservableList<>s, that's why I didn't connect that the output of .AsObservableList() isn't itself an IObservable<> that you can apply .DisposeMany() to.

As far as .DisposeMany() goes then, you can just move it up to right before the .AsObservableList() operator. As of 8.3.25, .DisposeMany() can be placed anywhere in the stream, where the items to be disposed are accessible, because it guarantees that all downstream notifications have been published before it performs disposal. I.E. The items will be fully-removed from the materialized IObservableList<>, before .Dispose() is called. The one exception being if there is a context switch downstream, E.G. a downstream .ObserveOn() call that marshals notifications to another thread. In that case, it's possible for items to be disposed before all lists that have captured them receive a notification to remove them, but even then, the only issue is if you try and USE one of those items in a way that .Dispose() doesn't allow, within the time it takes for that Removal to be marshalled and processed.

However, looking at this version of .Switch(), even that shouldn't be necessary. When a switch occurs, the operator should be publishing Remove changes for all the items in the list at the time of the switch, so just a single .DisposeMany() downstream of the .Switch() would be able to catch all items. The only thing left to dispose is the inner list itself, and the .Switch() together with .AsObservableList() will take care of that, automatically. At least, that's what should be happening. If you have reason to believe it's not, there may be a bug buried in one of these operators.

What I actually want

Alright, so something like this...

public record MyDataItem
{
    public int Id { get; init; }
    public string Name { get; init; }
    public string GroupName { get; init; }
}

public record MyDataState
{
    public ImmutableList<MyDataItem> Items { get; init; }
}

public class MyDataStateService
{
    public IObservable<MyDataState> StateOverTime();
}

And what you want in the UI is something like...

public class MyMainModel
{
    public ObservableCollection<MyItemGroupModel> Groups { get; }
}

public class MyItemGroupModel
{
    public string GroupName { get; }
    public ObservableCollection<MyItemViewModel> Items { get; }
}

public class MyItemViewModel
    : INotifyPropertyChanged
{
    public int Id { get; }
    public string Name { get; }
}

Does this reasonably represent your scenario?

[glen-nicol]

Yep! That is a good model to use as an example. Thanks for taking the time to put that together!

On the topic of DisposeMany, perhaps we're no longer in sync. I thought that disposed the items in the stream not the stream itself. But The thing I need to dispose of is the stream via the list/cache that wraps it. Once again I realize my example above doesn't actually apply. Maybe lets come back to that after we cover how to properly connect your example together?

[JakenVeina]

I'm on my phone, at the moment; for writing out example code, I'd want to be at my desktop. This question I can answer, though.

I thought that disposed the items in the stream not the stream itself.

Yes, it disposes all items in the collection. Specifically, it disposes items when they get removed, AND when the stream is torn down, it disposes all items that have not yet been removed.

Maybe I'm missing something in your scenario, but you don't really NEED an operator to dispose "the stream itself". The stream handles its own teardown when a completion or error comes from upstream, or when the downstream subscription is disposed. There shouldn't be anything extra you need to do to clean up the stream.

[JakenVeina]

Maybe I'm missing something in your scenario

I was indeed missing something. .AsObservableList() returns an IObservableList<> which is IDisposable, so yeah, you do potentially need to handle disposal of that. Practically speaking, you could just... not. Disposing them won't really do anything important, the .Switch() will take care of cleanup of the subscriptions, and that's all that's really important.

But the truth is, it doesn't even matter. Because you can, in fact, just get rid of .AsObservableList() completely there, and then nothing disposable gets created in the first place

service
    .StateOverTime()
    .Select(t => CreateVms(t)
        .ToObservable()
        .ToObservableChangeSet())
    .Switch()
    .AddKey(x => x.Key)
    .AsObservableCache();

.ToObservableChangeSet() here returns IObservable<IChangeSet<T>>, so, the call to .Switch() gets made for IObservable<IObservable<IChangeSet<T>> which is, in fact, the exact same .Switch() method that .Switch(IObservable<IObservableList<T>>) calls internally.

[glen-nicol]

I think I realized the same thing late yesterday when I found the overload of Switch you mentioned. Just a journey of finding new operators. Coming from plain RX I find myself thinking in those operators and am missing how to make a cohesive solution in dynamic data. But it seems like its getting better with your help! Thanks!

[JakenVeina]

Yep! That is a good model to use as an example.

So, the first and most obvious thing I'd recommend is breaking apart your state model. I've tried version of a fully-centralized state (that seems to be what you're going for? with MyDataState?) model in a few different apps, and I've found that it gets unreasonable REALLY fast. Also, in a practical sense, there's exactly one real-world application I'm aware of that uses a fully-centralized immutable state model is actually Visual Studio. For them, having a unified centralized state model makes a lot of sense, because it's what enables very-efficient parallelization of all the analysis and processing of those files. Roslyn can spin of a number of different threads to process different pieces of the tree, and they can all operate fully-independently, because the state tree is guaranteed not to change. Even if the user changes a file while a compilation is happening, all VS has to do is tell processing threads to cancel, and start over, with the new state tree. Bottom line, fully-centralized state, in my opinion, is only useful in very specific scenarios.

To be more specific, I'm saying that if you're gonna use Dynamic Data, pull stuff that you intend to use with Dynamic Data out of the state tree, and it becomes WAY easier to manage.

public class MyDataStateService
{
    public IObservable<MySimpleDataState> StateOverTime();

    public ISourceCache<int, MyDataItem> ItemsByKey { get; }
}

Reduce MySimpleDataState to just plain-object kinda stuff, no collections of items. Possibly even split it up further, into chunks of related-state for different functional modules.

If this really isn't feasible for you, then your goal is to mimic this setup as closely as possible, I.E. get those items into an ISourceCache as far upstream as possible. There's a relatively-easy way to do this in DD:

var itemsByKey = stateService
    .StateOverTime()
    .Select(state => state.Items)
    .DistinctUntilChanged()
    .EditDiff(item => item.Id)
    .AsObservableCache(); // No need for this if you're only going to have one downstream subscriber.

The .EditDiff() operator automatically diffs entire collections, by key, into specific changesets. It's just horribly inefficient at scale, because it has to iterate the entire incoming collection for every change. This is why you can save a LOT of performance by just having your items be stored in an ISourceCache<> to start with.

Now, we can build a much more straightforward DD stream to group up items and build VMs.

public sealed class MyMainModel
    : IDisposable
{
    public MyMainModel(IObservableCache<int, MyDataItem> itemsByKey)
    {
        itemsByKey
            .Connect()
            .Group(item => item.GroupName)
            .Transform(group =>new MyItemGroupModel(group))
            .DisposeMany()
            .Bind(out var Groups)
            .Subscribe()
            .DisposeWith(Subscriptions);
    }

    public ObservableCollection<MyItemGroupModel> Groups { get; }
}

The .Group() operator will handle all the heavy lifting for you. If an item's GroupName changes, the group with the old GroupName value will receive a Remove change for that item, and the group with the new value will receive a corresponding Add. If a group becomes empty, the bound collection will receive a Remove for the entire group object.

Your situation might be a little more complex than just grouping on a single string value, but you can easily create your own custom type to act as the group key, if there are multiple bits of data you want to combine together to be accessible at the whole-group level. As long as the custom type implements IEquatable<T> for itself, with equality based on the key value or values you want, .Group() will handle it just fine.

Let's expand on what MyItemGroupModel does.

public sealed class MyItemGroupModel
    : IDisposable
{
    public MyItemGroupModel(IGroup<MyDataItem, int, string> group)
    {
        GroupName = group.Key;
        
        group.Cache
            .Connect()
            .DistinctValues(item => item.Id)
            .Transform(itemId => new MyItemViewModel(itemId, group.Cache))
            .DisposeMany()
            .Bind(out Items)
            .Subscribe()
            .DisposeWith(Subscriptions);
    }

    public string GroupName { get; }
    public ObservableCollection<MyItemViewModel> Items { get; }
}

Here, I'm making a point to avoid some VM churn by ignoring all changes except add and remove of unique ID values, I.E. VMs only get constructed when an item is first added, and only get removed when the item is removed, as keyed by ID. Item refreshes and replaces along the way get handled within the VM itself, as follows:

public sealed class MyItemViewModel
    : INotifyPropertyChanged,
        IDisposable
{
    public MyItemViewModel(
        int                             id,
        IObservableCache<MyDataItem>    items)
    {
        Id = id;
        
        items
            .Watch(id)
            .Select(change => change.Current.Name)
            .DistinctUntilChanged()
            .Subscribe(name => Name = name)
            .DisposeWith(Subscriptions);
    }

    public int Id { get; private init; }
    public string Name { get; private set; }
}

I'll leave the implementation of INotifyPropertyChanged to the reader ;).

This isn't even the only possibility. Maybe it makes more sense to you to do the grouping AFTER you've transformed to VMs. In fact, that would probably produce less VM churn, if GroupName changes frequently.

public sealed class MyMainModel
    : IDisposable
{
    public MyMainModel(IObservableCache<int, MyDataItem> itemsByKey)
    {
        itemsByKey
            .Connect()
            .DistinctValues(item => item.Id)
            .Transform(itemId => new MyItemViewModel(itemId, itemsByKey))
            .GroupOnObservable(item => item.WhenAnyValue(item => item.GroupName))
            .Transform(group =>new MyItemGroupModel(group))
            .DisposeMany()
            .Bind(out var Groups)
            .Subscribe()
            .DisposeWith(Subscriptions);
    }

    public ObservableCollection<MyItemGroupModel> Groups { get; }
}

With this, MyItemGroupModel remains unchanged, and MyItemViewModel just needs to add a GroupName property that publishes change notifications, the same way that Name does.

[glen-nicol]

I appreciate the suggestion, but I plan to keep the centralized state container. Its a Redux-ish store for coherency.

Your example makes sense and is more or less what I did in my first attempt (Except mine was quite messy with more operators. So thanks for showing me how to clean that up.) There is one main difference that I didn't catch in the hypothetical example you presented to model my situation.

Lets say there is more information associated with the group. An enum to determine how its grouped, and then the group name is determined by the collection of items.

Here are the lines in particular that tripped my first attempt up

I created a single record GroupHeaderViewModel(Enum GroupKey, string HeaderTitle) that was given to each related item (doable because I have all the state at once up front). And then I grouped on that header. Finally I used the IGrouping.Key like you did to present the title for data binding and sorting on MyItemGroupModel and then the group items were pumped into an ObservableCollection.

The problem is the title messes up the keys partitioning so the groups go wacky as they change. I tried to take it out of the key .GroupBy(i => i.Group.Key) but then I can't access the title immediately from IGrouping.Key when creating the MyItemGroupModel. So
my solution to that was to make the title on MyItemGroupModel an observable property that is based on the items in the group. But once I did that, then sorting threw an NRE because it sorted before the items hydrated and there was no header information. Obviously I can make the sorting smarter to handle nulls but it just make we wonder if there is an easier/prescribed way?

[JakenVeina]

The problem is the title messes up the keys partitioning so the groups go wacky as they change.

Yeah, most of the grouping operators do not support allowing the group key to change. .GroupOnObservable() is what you need, it will safely handle re-grouping an item when it's group key changes, whether that's a single value, or a composite. You do need to treat key values as immutable, though, I.E. if your group key is an object, and you want to change one of its properties, you need to swap out the whole object for a new one.

[glen-nicol]

for sure, I guess I will consider my case unique then and make some custom operator for it. Thanks for all your help!

[JakenVeina]

A custom operator might be nice if you have a lot of different streams like this in different places in your app, but from what I can see, you should be able to do this exclusively with DD operators.