反应式扩展和 IDisposable 管理

Question

我正在尝试使用 Microsoft.Management.Infrastructure 的 Async 版本，但我很难弄清楚如何使用 Reactive Extensions 进行会话管理。

这里有一些背景：

• 可以使用CimSession.CreateAsync 创建一个CIM 会话，返回IObservable
• CimSession.CreateAsync 返回一个 CimAsyncResult<T>，它包裹了 CimAsyncDelegatedObservable<T>
• CimAsyncDelegatedObservable<T>does not dispose CimSession 在任何时候，因此必须在不再需要 CimSession 时手动处理它。
• 获得CimSession 的句柄后，我尝试使用QueryInstancesAsync 运行WQL 查询，例如session.QueryInstancesAsync(@"root\cimv2", "WQL", "SELECT Username FROM Win32_ComputerSystem")

如果CimSession 返回Tasks 而不是Observables，代码将非常简单：

using var session = CimSession.CreateAsync(".");
var results = session.QueryInstancesAsync(@"root\cimv2", "WQL", "SELECT Username FROM Win32_ComputerSystem");

但是由于Async 方法是使用Observables 实现的，我不确定如何将其转换为Observables 的惯用用法。当然，我可以使用 System.Reactive.Linq.Observable.ForEachAsync 将我的方法转换为 Tasks，但我想学习如何正确使用 Observables。

总结一下，我的问题是：

• 将CimSession 的生命周期绑定到Observable
• 如果没有订阅者，如何确保CimSession 得到处理？我应该强制至少一个订阅者，但有工厂方法吗？
• 当QueryInstancesAsync 的结果到达时如何处理CimSession observable？

Answer 1

这是一个有点古怪的界面，但试试这个，让我知道它是否有效：

IObservable<string> query =
    from session in CimSession.CreateAsync(".")
    from x in Observable.Using(
        () => session,
        s => s.QueryInstancesAsync(@"root\cimv2", "WQL", "SELECT Username FROM Win32_ComputerSystem"))
    from y in Observable.Using(
        () => x,
        z => Observable.Return(z.GetCimSessionComputerName()))
    select y;

---

如果可行，我建议将其包装在一个单独的类中：

public static class CimSessionEx
{
    public static IObservable<T> CreateObservable<T>(string computerName, string namespaceName, string queryDialect, string queryExpression, Func<CimInstance, IObservable<T>> factory) =>
        from session in CimSession.CreateAsync(computerName)
        from instance in Observable.Using(() => session, s => s.QueryInstancesAsync(namespaceName, queryDialect, queryExpression))
        from y in Observable.Using(() => instance, factory)
        select y;
}

那么你可以这样使用它：

IObservable<string> query =
    CimSessionEx.CreateObservable(
        ".",
        @"root\cimv2",
        "WQL",
        "SELECT Username FROM Win32_ComputerSystem",
        i => Observable.Return(i.GetCimSessionComputerName()));