Is it possible to use the Using operator in Rx.Net with a resource that implements IAsyncDisposable
rather than IDisposable
? If not, is there some sort of workaround that I could use?
CodePudding user response:
Here is a Using
method that works with IAsyncDisposable
objects:
/// <summary>
/// Constructs an observable sequence that depends on a resource object,
/// whose lifetime is tied to the resulting observable sequence's lifetime.
/// </summary>
public static IObservable<TResult> Using<TResult, TResource>(
Func<TResource> resourceFactory,
Func<TResource, IObservable<TResult>> observableFactory)
where TResource : IAsyncDisposable
{
return Observable.Defer(() =>
{
TResource resource = resourceFactory();
IObservable<TResult> observable;
try { observable = observableFactory(resource); }
catch (Exception ex) { observable = Observable.Throw<TResult>(ex); }
Lazy<Task> lazyDisposeTask = new(() => resource.DisposeAsync().AsTask());
IObservable<TResult> disposer = Observable
.FromAsync(() => lazyDisposeTask.Value)
.Select(_ => default(TResult))
.IgnoreElements();
return observable
.Catch((Exception ex) => disposer.Concat(Observable.Throw<TResult>(ex)))
.Concat(disposer)
.Finally(() => lazyDisposeTask.Value.GetAwaiter().GetResult());
});
}
This method has identical signature with the Rx Observable.Using
method (apart from the where
clause), and it can be used in the same way.
This implementation takes care of all completion cases:
- Successful completion: The
IAsyncDisposable
resource is disposed asynchronously by theConcat
operator. - Completion with error: The
IAsyncDisposable
resource is disposed asynchronously by theCatch
operator. - The sequence is unsubscribed before its completion: The
IAsyncDisposable
resource is disposed synchronously by theFinally
operator. Disposing asynchronously the resource is not possible in this case, for reasons explained here.
Variant with asynchronous factory methods:
public static IObservable<TResult> Using<TResult, TResource>(
Func<CancellationToken, Task<TResource>> resourceFactoryAsync,
Func<TResource, CancellationToken, Task<IObservable<TResult>>> observableFactoryAsync)
where TResource : IAsyncDisposable
{
return Observable.Create<TResult>(async (observer, cancellationToken) =>
{
TResource resource = await resourceFactoryAsync(cancellationToken);
IObservable<TResult> observable;
try { observable = await observableFactoryAsync(resource, cancellationToken); }
catch { await resource.DisposeAsync(); throw; }
Lazy<Task> lazyDisposeTask = new(() => resource.DisposeAsync().AsTask());
IObservable<TResult> disposer = Observable
.FromAsync(() => lazyDisposeTask.Value)
.Select(_ => default(TResult))
.IgnoreElements();
return observable
.Catch((Exception ex) => disposer.Concat(Observable.Throw<TResult>(ex)))
.Concat(disposer)
.Finally(() => lazyDisposeTask.Value.GetAwaiter().GetResult())
.Subscribe(observer);
});
}