Me gustaría crear un método usando rxjs para consultar un punto final, que devuelve una matriz de objetos, luego usar datos de esa respuesta para obtener detalles adicionales sobre el elemento en la matriz desde otro punto final y finalmente devolver una única matriz de objetos como un observable.
Este código funciona, pero me gustaría hacerlo sin tener una suscripción dentro del operador del grifo.
testMethod() { this.testService.getItemList().pipe( tap((items) => { items.forEach((item, i) => { this.itemArr.push(item); this.testService.getItemDetails(item.id).subscribe(itemDetails => { this.itemArr[i]['details'] = itemDetails; }); }); }) ).subscribe(); }Hay varias formas de hacer algo como esto con rxJs. Varían en la forma en que gestionan el paralelismo de las llamadas al segundo extremo, es decir, el extremo que devuelve detalles por cada elemento.
MÁXIMO PARALELISMO
Supongamos que recibe 10 elementos en la matriz desde el primer punto final y desea ejecutar todas las 10 llamadas al segundo punto final en paralelo. En este caso, puede usar el operador forkJoin de esta manera
testMethod() { this.testService.getItemList().pipe( // transform the array of items into an array of Observables // note that the outer map is the rxJs operator while the inner one is the // the javascript array method // note also that we return the array of items since we will need it later map(items => [items.map(item => this.testService.getItemDetails(item.id)), items] ), // then switch to a new Observable which will emit when all of the calls // to the second endpoint have returned switchMap(([arrayOfObs, items]) => forkJoin(arrayOfObs).pipe( // return both the results of the calls to the second endpoint and the // original array of items map(itemDetails => [itemDetails, items]) )), // finally augment the original array of items with the detail info // as in your original code but with no subscription any more tap(([itemDetails, items]) => { items.forEach((item, i) => { this.itemArr.push(item); this.itemArr[i]['details'] = itemDetails[i]; }); }) ).subscribe(); }SIN PARALELISMO
Si desea ejecutar las llamadas al segundo punto final secuencialmente, puede usar el operador concatMap de esta manera
testMethod() { this.testService.getItemList().pipe( // transform the array of items into a new stream which notifies sequentially // each item in the array - we use the from rxJs function to create the new stream (ie the new Observable) switchMap(items => from(items)), // then concatenate the calls to the second endpoint with concatMap concatMap(item => this.testService.getItemDetails(item.id).pipe( // return the original item with its details map(itemDetail => { item['details'] = itemDetail; return item }) )), // finally gather all items into an array toArray() ).subscribe(); }CONCURRENCIA CONTROLADA
Si desea un cierto nivel de paralelismo, por ejemplo, 5 llamadas en paralelo como máximo, puede sustituir contactMap con mergeMap especificando el nivel de concurrencia usando el segundo parámetro de mergeMap .
Tu solución funciona bien. Sin embargo, creo que el código se vuelve un poco más simple si primero emite los elementos devueltos uno a la vez, lo que puede lograr comenzando con switchMap(items => items) :
testMethod(): Observable<ItemWithDetails[]> { return this.testService.getItemList().pipe( switchMap(items => items), mergeMap(item => this.testService.getItemDetails(item.id).pipe( map(details => ({...item, details})) )), reduce((acc, curr) => [...acc, curr], []) ); } El switchMap se ve un poco extraño, pero en realidad solo emite los elementos de la matriz uno a la vez ( piense en ello como switchMap(items => from(items)) . Dado que from usa internamente, no es necesario que lo incluyamos). .
mergeMap toma el elemento y realiza la llamada para obtener detalles, luego devuelve un objeto con los detalles adjuntos.
reduce acumula los resultados y los emite una vez que se completa el observable. ( Si está interesado en emitir a medida que se reciben los resultados, use el scan en su lugar. Esto puede ser útil cuando hay muchos datos y desea actualizar la interfaz de usuario con los datos a medida que se reciben )
Para hacerlo aún más claro, podría dividir la parte que agrega los detalles y terminar con esto:
testMethod(): Observable<ItemWithDetails[]> { return this.testService.getItemList().pipe( switchMap(items => items), mergeMap(item => this.appendItemDetails(item)), reduce((acc, curr) => [...acc, curr], []) ); } private appendItemDetails(item: Item): Observable<ItemWithDetails> { return this.testService.getItemDetails(item.id).pipe( map(details => ({...item, details})) ); }Pude hacer que funcionara usando una combinación de los operadores mergeMap, merge y reduce.
testMethod() { this.testService.getItemList().pipe( mergeMap(items => { const item = items.reduce((acc, curr) => { const itemArr = this.testService.getItemDetails(curr.id).pipe( map(items => ({...curr, items})) ); return [...acc, itemArr]; },[]); return merge(...item); }), reduce((acc, curr) => ([...acc, curr]),[]) ); }