是否可以将&Quot；运算符与IAsyncDisposable配合使用？

本文介绍了是否可以将&Quot；运算符与IAsyncDisposable配合使用？的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述 是否可以对实现IAsyncDisposable而不是IDisposable的资源使用rx中的Using操作符？如果没有，是否有我可以使用的解决方法？

推荐答案

这里有一个Using方法，可以处理IAsyncDisposable对象：

/// <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()); }); }

此方法与RxObservable.Using方法具有相同的签名(除了where子句)，并且可以以相同的方式使用。

此实现处理所有完成案例：

成功完成：IAsyncDisposable资源由Concat运算符异步释放。

完成但出错：IAsyncDisposable资源由Catch运算符异步释放。

序列在完成前取消订阅：IAsyncDisposable资源由Finally运算符同步处理。在这种情况下，异步处置资源是不可能的，原因如下here。

带有异步工厂方法的变量：

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); }); }