javascript - mergeall - rxjs merge

RxJS: concat tres promesas, distinguen resultados

Question

Tengo tres promesas, las solicitudes de Rest devuelven listas de datos.
El tercero tiene referencias (ids) a las dos primeras listas, por lo que quiero asignar estos identificadores a los nombres correspondientes cuando tenga todos los datos.
El mapeo no es un problema, solo uso Lodash para eso.
Pero el problema es esperar a que se resuelvan las tres promesas antes de comenzar a calcular este mapeo.

Me di cuenta de usar concat() :

Rx.Observable.concat(p1, p2, p3).subscribe(
function onNext(list)
{
  // Assign the list to the corresponding variable in the scope
},
function onError(e)
{
  // Notify of error
},
function onCompleted()
{
  // Do the mapping
}
);

Mi problema es que se onNext() en orden aleatorio. Es decir. No sé qué lista recibiré en algún momento, y es difícil saber a partir de los datos que contiene.

¿Hay alguna manera de rastrear qué promesas produjo qué lista? Una especie de zip? concatMap? concatMapObserver? Admito que no he entendido completamente el uso de los últimos dos ...

Answer 1

Hice algunos experimentos esta mañana, antes de ir a ver si tengo una respuesta ... :-)

Puedes verlos en http://plnkr.co/edit/3Xczzw

Descubrí que el concat de observables se ejecutará cada uno por turno, en el orden dado.

  var o1 = Rx.Observable.interval(1500).take(1).map(function (i) { return { n: i, id: ''First'', ts: Date.now() - reference }});
  var o2 = Rx.Observable.interval(1000).take(2).map(function (i) { return { n: i, id: ''Second'', ts: Date.now() - reference }});
  var o3 = Rx.Observable.interval(2000).take(1).map(function (i) { return { n: i, id: ''Third'', ts: Date.now() - reference }});
  // Alternative
  var oa = Rx.Observable.timer(1200).map(function (i) { return { n: i, id: ''Alternative'', ts: Date.now() - reference }});

  Rx.Observable.concat(o1, o2, o3, oa).subscribe(
      function onNext(v) { v.timestamp = Date.now() - reference; showObject(v); },
      function onError(e) { var ts = Date.now() - reference; showHTML("Error " + JSON.stringify(e) + " at " + ts); },
      function onCompleted() { var ts = Date.now() - reference; showHTML("Completed at " + ts); }
  );

da

{"n":0,"id":"First","ts":1503,"timestamp":1503}
{"n":0,"id":"Second","ts":2504,"timestamp":2504}
{"n":1,"id":"Second","ts":3505,"timestamp":3505}
{"n":0,"id":"Third","ts":5506,"timestamp":5506}
{"n":0,"id":"Alternative","ts":6708,"timestamp":6708}
Completed at 6708

Un concat de promesas no entregará nada antes de que se resuelva la primera promesa. Luego puede entregar (es decir, llamar a Siguiente) las otras promesas resueltas, aún en el orden dado. Entonces puede esperar la próxima promesa si queda algo, etc.

  var p1 = promiseInTime(1500, { id: ''First''});
  var p2 = promiseInTime(1000, { id: ''Second'' });
  var p3 = promiseInTime(2000, { id: ''Third'' });
  var pa = promiseInTime(1200, { id: ''Failed? '' + !!withFailure }, withFailure);

  Rx.Observable.concat(p1, p2, p3, pa).subscribe(
      function onNext(v) { v.timestamp = Date.now() - reference; showObject(v); },
      function onError(e) { var ts = Date.now() - reference; showHTML("Error " + JSON.stringify(e) + " at " + ts); },
      function onCompleted() { var ts = Date.now() - reference; showHTML("Completed at " + ts); }
  );

da

{"id":"First","promiseTimeout":1500,"timestamp":1501}
{"id":"Second","promiseTimeout":1000,"timestamp":1506}
{"id":"Third","promiseTimeout":2000,"timestamp":2001}
Error {"id":"Failed? true","promiseTimeout":1201} at 2004

o

{"id":"First","promiseTimeout":1500,"timestamp":1501}
{"id":"Second","promiseTimeout":1000,"timestamp":1503}
{"id":"Third","promiseTimeout":2000,"timestamp":2000}
{"id":"Failed? false","promiseTimeout":1201,"timestamp":2004}
Completed at 2004

Entonces, básicamente, concat respeta el orden de sus argumentos, lo cual puede ser una forma de encontrar qué pedido emitió un resultado de promesa.
En el caso de las solicitudes de Ajax, mejor concat las promesas que las observables, ya que se solicitarán en paralelo, no de forma secuencial (a menos que necesites la última, por supuesto).

Intenté con la solución proporcionada por @ user3743222, y es una buena solución. Prefiero forkJoin, ya que los resultados se asignan explícitamente a los parámetros, en lugar de confiar en el orden.

Debe tener en cuenta las diferencias en la gestión de errores: descubrí que concat procesará todas las promesas hasta que se encuentre el primer error. Mientras forkJoin no procesará nada si uno de los promesas comete errores. Lo cual tiene sentido (no podemos unir resultados parciales, en general).

Answer 2

Si se trata de promesas de las que estamos hablando, creo que puede echarle un vistazo al operador forkJoin . Cf. https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/forkjoin.md , http://xgrommx.github.io/rx-book/content/observable/observable_methods/ forkjoin.html

Entonces podrías tener algo como:

Rx.Observable.forkJoin(p1, p2, p3, function(p1,p2,p3){
/* your combining code here */
})

En resumen, forkJoin tiene una semántica similar a la de RSVP. RSVP.all si conoce las promesas de la biblioteca RSVP.