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

Question

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

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

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

Observable 都是 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();
            });
    });

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

Answer 1

与其将代码放入 Subscribe 处理程序，不如将其作为序列的一部分。您可以使用 Select 运算符将每个 listJson 投影到 IObservable<string>（产生嵌套的 IObservable<IObservable<string>>），然后使用 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 将导致对 processListObservable 的隐式订阅，您的代码将作为此订阅的副作用执行。

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

How to combine IObservable sequences in UniRx/Rx.NET?

c#

unity3d

reactivex

rx.net

unirx