将 Observable<List<X>> 转换为 Observable<X> 的 rxjs 运算符

Question

我正在使用 Angular HTTP 和 rxjs 处理一个用户友好的 API。我正在尝试从我的服务中返回Observable<List<Object>>。下面是我的 get 请求返回的 JSON 示例：

Items: {
​​    "$type": "System.Collections.Generic.List`1[[Asi.Soa.Core.DataContracts.GenericEntityData, Asi.Contracts]], mscorlib"
​​    "$values": [
       Object { "$type": "Asi.Soa.Core.DataContracts.GenericEntityData, Asi.Contracts", EntityTypeName: "7", Properties: {…} }
       Object { "$type": "Asi.Soa.Core.DataContracts.GenericEntityData, Asi.Contracts", EntityTypeName: "7", Properties: {…} }
       Object { "$type": "Asi.Soa.Core.DataContracts.GenericEntityData, Asi.Contracts", EntityTypeName: "7", Properties: {…} }
    ]
}

然后在每个Object里面：

{
  "$type": "Asi.Soa.Core.DataContracts.GenericEntityData, Asi.Contracts"
   EntityTypeName: "7"
   Properties: {
    "$type": "Asi.Soa.Core.DataContracts.GenericPropertyDataCollection, Asi.Contracts"
    "$values": {}
}

我需要为每个项目提取$values，然后返回$values 对象的列表。但是我不知道如何在没有订阅的情况下使用 Angular 和 rxjs 做到这一点。而且我不想订阅，因为我希望 组件，而不是 服务，来处理结果数据。

我一直在寻找 rxjs Operators 来解决我的问题。但是似乎没有一个运算符可以将 List<X> 转换为 X 同时保留 Observable。

这可能与 Angular2+ 和 rxjs 有关吗？或者是否有更好的方法来完成这项任务。我对响应式编程比较陌生，所以任何一般性的建议都将不胜感激。

Answer 1

你在 RxJS 中使用 map()-operator 转换元素。它将更改应用于流中的每个元素并返回更改后的元素。

subscribe() 的目的是首先触发所有 RXJS 函数。因为即使一个冷的 observable 获得了一个新的值，它也不会发生任何事情，除非它最后都是订阅。所以你可以做

const sub = $stream.map(originalValue => {return transformToX(originalValue);})
.subscribe(transformedValue => {
    emitter.emit(transformedValue);
});

或者更清楚地展示订阅的目的：

const sub = $stream.map(originalValue => {return transformToX(originalValue);})
.do(transformedValue => {
    emitter.emit(transformedValue);
})
.subscribe();

当您不再需要所有订阅时，不要忘记取消订阅！请参阅http://brianflove.com/2016/12/11/anguar-2-unsubscribe-observables/。

Answer 2

使用Observable#mergeMap（在其他实现中通常称为flatMap）。

mergeMap 允许您：

1. 将您的每个项目映射到列表中（即，为每个事件发出一个值列表）
2. 扁平化将列表重新组合成一个事件流

---

假设收到以下数据：

const result = {
  items: {
    $type: 'something something',
    $values: [1, 2, 3]
  }
}

您可以使用以下方法获取其值的Observable：

Rx.Observable.of(result).mergeMap(x => x.items.$values)

例如：

Rx.Observable
  .of(result)
  .mergeMap(x => x.items.$values)
  .subscribe(value => console.log(`new value: ${value}`));

... 会打印出来：

新值：1
新值：2
新值：3