arrays - 如何使用响应式(Reactive)扩展将一个事件拆分为多个事件？

Question

如何将响应式(Reactive)扩展流中的单个事件拆分为同一流中的多个事件？

我有一个检索 json 数据的序列，它是顶层的一个数组。在解码 json 数据时，我想获取该数组中的每个元素并继续沿流传递这些元素。

这是一个我希望存在的虚构函数的示例(但名称更短!)。它是用 Python 编写的，但我认为它足够简单，其他 Rx 程序员应该也能看懂。

# Smallest possible example
from rx import Observable
import requests
stream = Observable.just('https://api.github.com/users')
stream.map(requests.get) \
      .map(lambda raw: raw.json()) \
      .SPLIT_ARRAY_INTO_SEPARATE_EVENTS() \
      .subscribe(print)

换句话说，我想像这样进行转换:

From:
# --[a,b,c]--[d,e]--|->
To:
# --a-b-c-----d-e---|->

Answer 1

您可以使用 SelectMany 运算符:

stream.SelectMany(arr => arr)

这将“扁平化”您的事件流，就像 C# LINQ SelectMany 运算符可用于扁平化一系列序列一样。