如何在 UniRx/Rx.NET 中组合 IObservable 序列？

如何解决如何在 UniRx/Rx.NET 中组合 IObservable 序列？

我正在将 UniRx 风格的 Reactive Extensions 用于 Unity3D 游戏引擎。 Unity 使用 C#，所以我猜它类似于 Rx.NET。

我需要一种更漂亮的方法来检查几个可观察序列何时完成。

在下面的示例中，其中一个序列取决于第一个的结果（因为它需要一个整数来表示 processID）。

observables 都是 IObservable<string> 类型。

var processListObservable = APIBuilder
    .GetProcessList(authInfo.Token,authInfo.PlatformURL,(int)product.Id)
    .Subscribe(listJson =>
    {
        processList = ProcessList.FromJson(listJson);
        int processID = (int)processList.Processes[0].ProcessDataId;

        //Retrieve Detailed information of the first entry
        var processDetailsObservable = APIBuilder
            .GetProcessDetails(token,platformURL,product.Id,processID)
            .Subscribe(detailsJson =>
            {
                processData = ProcessData.FromJson(detailsJson);
                SetupPlotView();
            });
    });

任何提示将不胜感激。还有一些解决相同场景的建议，减去对第一个序列结果的依赖。

解决方法

与其将您的代码放入 Subscribe 处理程序，您还可以使其成为序列的一部分。您可以使用 Select 运算符将每个 listJson 投影到 IObservable<string>（导致嵌套的 IObservable<IObservable<string>>），然后使用 {{ 1}} 或 Concat 运算符，具体取决于您要阻止还是允许并发。

Merge

最后一行中的 var processListObservable = APIBuilder .GetProcessList(authInfo.Token,authInfo.PlatformURL,(int)product.Id) .Select(listJson => { var processList = ProcessList.FromJson(listJson); int processID = (int)processList.Processes[0].ProcessDataId; return APIBuilder.GetProcessDetails(token,platformURL,product.Id,processID); }) .Concat() // or .Merge() to allow concurrency .ObserveOn(SynchronizationContext.Current) // Optional .Do(detailsJson => { var processData = ProcessData.FromJson(detailsJson); SetupPlotView(processData); }); await processListObservable.DefaultIfEmpty(); // Start and await the operation 将导致对 await 的隐式订阅，并且您的代码将作为此订阅的副作用执行。