rxjs Operator that converts Observable<List<X>> to Observable<X>

Question

I'm working with a no-so user-friendly API using Angular HTTP and rxjs. I'm trying to get return an Observable<List<Object>> from my service. Here's a sample of the JSON that's returned from my get request:

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: {…} }
    ]
}

And then inside each Object:

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

I need to extract out $values for each item, and then return a list of the $values objects. But I can't figure out how to do that with Angular and rxjs without subscribing. And I don't want to subscribe, as I want the component, not the service, to process the resulting data.

I've been looking in rxjs Operators as a solution to my problem. But there doesn't seem to be an operator that converts a List<X> into X while preserving the Observable.

Is this possible to do with Angular2+ and rxjs? Or is there maybe a better way to accomplish this task. I'm relatively new to reactive programming, so any general advice would be appreciated.

Answer 1

Use Observable#mergeMap (often called flatMap in other implementations).

mergeMap allows you to:

1. map each of your items into lists (that is, emit a list of values for each event)
2. flatten the lists together back into one stream of events

---

Assuming the following data received:

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

You could get an Observable of its values using the following:

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

For instance:

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

... Would print out:

new value: 1
new value: 2
new value: 3

Answer 2

You transform elements in RxJS with the map()-operator. It applies a change to every element in the stream and returns the changed one.

The purpose of subscribe() is to make all the RXJS-functions being triggered in the first place. Because even when a cold observable gets a new value, nothing happens to it unless at the end of it all is a subscription. So you can do

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

Or to demonstrate the purpose of subscribe more clearly:

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

Do not forget to unsubscribe from all subscribtions when you don't need them anymore! See http://brianflove.com/2016/12/11/anguar-2-unsubscribe-observables/ for that.