[Feature]: Support DisposeManyAsync

[Dunge]

Describe the functionality desired 🐞

Reopening #404 since it was closed and not completed and is still relevant.

I have a connected ObservableCollection that the objects would require to implement IAsyncDisposable.
I managed to get it working y wrapping it in a Task.Run() inside the normal Dispose, but that doesn't seem safe.

Sorry I don't have a PR to provide, just opening a ticket so that it remains in the known list of stuff to do.

The steps the functionality will provide

?

Considerations

?

[JakenVeina]

IAsyncDisposable is pretty darn niche, but I see no reason not to support it, if we support IDisposable.

I will say the operator name should, I think, be AsyncDisposeMany(), or even DisposeAsyncMany(). Async() as a method suffix should be reserved exclusively for awaitable methods.

[Dunge]

Agreed for the name.

My use case if interested : I use DynamicData in a Blazor application to obtain a filtered/transformed derived collection based on another. The elements represents polylines on a google map component. Everything to interact with it is done via javascript interops, which are async calls. When an element gets removed from the collection, I need to do one and the dispose methods seems to be the correct place for that.

This lead me to another similar issue on creation. It would also be interesting to have the capabilities for an async Initialiser method since constructors can't be. I'm once again required to launch a task unawaited.. But this is not the scope of this ticket.

[JakenVeina]

@dwcullop @RolandPheasant

So, there's a couple ways we could support this, that I'm brainstorming.

1. A new operator .AsyncDisposeMany() that requires TObject : IAsyncDisposable
2. A new operator .AsyncDisposeMany() that (like .DisposeMany()) does not require TObject : IAsyncDisposable, and just does a type check on each item for whether or not it needs disposal, at the time of removal.
3. Since .DisposeMany() already doesn't require TObject : IDisposable we could just add support for IAsyncDisposable to it, internally, in the same fashion as IDisposable, but with the additional plumbing to ensure that the .DisposeAsync() calls get properly awaited and plumbed into the stream. I.E. errors get caught and published downstream, and completion of the stream waits for completion of all outstanding disposals.

The other concern that occurs to me, for all of the above options, is that if an un-subscription occurs from downstream, then outstanding disposals become orphaned. Like, we can still DO all the disposals, but any exceptions that occur would get swallowed, because there'd be no downstream to send them to anymore. With .DisposeMany() today, any exceptions occurring during un-subscription will throw up to the subscription.Dispose() caller, immediately. I can't think of any effective way around this that isn't "make the whole downstream IAsyncDisposable<>", or "just document on the operator that you shouldn't un-subscribe from downstream".

[RolandPheasant]

I like the idea of 3, but you'd have to work out how to invoke the async part from non-async code in a safe manner, so 2 may be better but let's see what you can come up with.

I am a bit vague on the original rx guideline by I seem to recall that the recommendation is never to throw an exception inside inside a dispose. I may be wrong however, but that said it would be dangerous to swallow an error.

[JakenVeina]

you'd have to work out how to invoke the async part from non-async code in a safe manner

Invoking async code within and RX query is trivial, that's not the problem. The problem is that RX is only async going downstream, not going upstream (I.E. un-subscription). The burden is on the consumer to safely do teardown of a DD query, if there's IAsyncDisposable objects involved, by not just un-subscribing.

I seem to recall that the recommendation is never to throw an exception inside inside a dispose

I mean, that's really a part of the contract of IDisposable in general, whenever possible. But propagation of errors is only one of the things that gets lost for un-awaited Tasks, the other is completion notifications. And really, situations where waiting for completion of teardown is important is kinda the entire reason that IAsyncDisposable exists as an interface.

[JakenVeina]

Yeah, this definitely has to be a new operator. I can still have that operator support both IDisposable and IAsyncDisposable, but the introduction of IAsyncDisposable throws some real wrenches into the safety of the operator. The main point being that we REALLY need to propagate completion of .DisposeAsync() calls, as I mentioned above. That becomes impossible to do within the primary stream, because an error within the stream will trigger disposals, and then, by RX rules, tears down the stream, meaning we CANNOT report on completion of the disposals. We have to have a second side-stream that carries disposal completion (and errors, if that should occur, although it really shouldn't).

public static IObservable<IChangeSet<TObject, TKey>> AsyncDisposeMany(
    this    IObservable<IChangeSet<TObject, TKey>>  source,
            IObserver<Unit>                         onDisposalsCompleted);

public static IObservable<IChangeSet<TObject, TKey>> AsyncDisposeMany(
    this    IObservable<IChangeSet<TObject, TKey>>  source,
    out     IObservable<Unit>                       disposalsCompleted);

That disposalsCompleted event can then be used to drive completion of .DisposeAsync() on the class that owns this query.

[JakenVeina]

@Dunge #1011 is the implementation I came up with. Does this work for you?

[Dunge]

@JakenVeina thank you very much! I assume it does, I would have to test it in a real world situation to see if it reacts as expected. You went slightly above my capacity expertise in the theory and implementation using streams to report the disposal completion, so unfortunately I can't comment on the safety and performance implications, but I'm sure it is perfect!

Now I don't want to push my luck, but could this recipe also be used to support an async initialization call? When using the SourceCache.Connect().Transform(t => new xxx()) pattern. I hope I'm not saying something that doesn't logically make sense here, I know constructors can't be async, but you often have an InitializeAsync() method to call after the constructor, using a factory or other mean to do it. Could this be wired up in the DynamicData pipeline in some way?

[JakenVeina]

Sure can. Something like...

using var source = new SourceCache<Item, long>(static item => item.Id);

Task? whenDisposalsComplete = null;

var items = source
    .Connect()
    .Transform(...)
    .AsyncDisposeMany(disposalsCompleted => whenDisposalsComplete = disposalsCompleted.ToTask())
    .AsObservableCache();

using var binding = items
    .Bind(out var boundItems)
    .Subscribe();

using var initializations = items
    .SelectMany(changes => changes
        .Where(change => change.Reason is ChangeReason.Add)
        .Select(change => Observable.FromAsync(change.Current.InitializeAsync)))
    .Merge()
    .Subscribe();

// Do stuff

initializations.Dispose();
binding.Dispose();
source.Dispose();

await whenDisposalsComplete;

There might be a slightly better way to get a stream of all added items, than that big .SelectMany(), but I'm not sure, off tge top of my head. The goal, really, is to take each item that gets added to the collection and merge it into a single result (in this case a merged stream), and in theory, what that would be is a custom .Aggregate() operation, so you may want to look at that option. I'm not sure if the verbosity would be worth it.

Actually, an entirely different thought just occurred to me, that probably makes way more sense...

using var source = new SourceCache<Item, long>(static item => item.Id);

Task? whenDisposalsComplete = null;

var items = source
    .Connect()
    .TransformOnObservable(item =>
    {
        var asyncDisposableItem = new AsyncDisposableItem()
        {
            ...
        };

        return Observable
            .FromAsync(asyncDisposableItem.InitializeAsync)
            .Select(_ => asyncDisposableItem);
    })
    .AsyncDisposeMany(disposalsCompleted => whenDisposalsComplete = disposalsCompleted.ToTask())
    .AsObservableCache();

using var binding = items
    .Bind(out var boundItems)
    .Subscribe();

// Do stuff

binding.Dispose();
source.Dispose();

await whenDisposalsComplete;

This approach treats .InitializeAsync() much more like a true async constructor: the items don't even ENTER the bound collection, and become subject to disposal, until after they are fully initialized.

[github-actions]

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