Pesquisa de site

javascript - RXJS Encadear observáveis dependentes sequencialmente, mas obtendo cada emissão para uma barra de progresso

Estou enfrentando um problema e tenho tentado encontrar uma solução usando RxJs, mas não consigo encontrar uma que se encaixe...

  • Eu tenho 3 solicitações REST diferentes, que serão chamadas sequencialmente, e cada uma delas precisa da resposta da anterior como argumento
  • Quero implementar uma barra de progresso, que aumenta à medida que os pedidos têm sucesso

Aqui está o que eu pensei :

  • Vou usar pipes e concatMap() para evitar assinaturas aninhadas e me inscrever em cada requisição quando a anterior estiver concluída.

Considere esta versão muito simplificada. Assuma que cada of representa uma requisição REST bem-sucedida completa (lidaremos com erros mais tarde), e que eu farei trabalhos não mostrados com o parâmetro n...

const request1 = of('success 1').pipe(
  delay(500),
  tap(n => console.log('received ' + n)),
);

const request2 = (n) => of('success 2').pipe(
  delay(1000),
  tap(n => console.log('received ' + n))
);

const request3 = (n) => of('success 3').pipe(
  delay(400),
  tap(n => console.log('received ' + n))
);

request1.pipe(
  concatMap(n => request2(n).pipe(
    concatMap(n => request3(n))
  ))
)

No entanto, quando me inscrevo na última parte do código, receberei apenas a resposta da última requisição, o que é esperado, já que o pipe resolve para isso.

Então, com concatMap(), eu posso encadear minhas chamadas REST dependentes corretamente, mas não consigo acompanhar o progresso.

Embora eu pudesse acompanhar o progresso com bastante facilidade usando inscrições aninhadas, estou me esforçando para evitar isso e usar a melhor forma prática.

Como posso encadear minhas chamadas REST dependentes, mas ainda ser capaz de fazer coisas cada vez que uma chamada tiver sucesso?

Resposta #1

Esta é uma solução generalizada, embora não seja tão simples. Mas ela torna o progresso observável enquanto ainda evita o operador share, que pode introduzir estado inesperado se usado incorretamente.

const chainRequests = (firstRequestFn, ...otherRequestFns) => (
  initialParams
) => {
  return otherRequestFns.reduce(
    (chain, nextRequestFn) =>
      chain.pipe(op.concatMap((response) => nextRequestFn(response))),
    firstRequestFn(initialParams)
  );
};

chainRequests recebe um número variável de funções e retorna uma função que aceita parâmetros iniciais e retorna um observable que concatMaps as funções juntas conforme mostrado manualmente na pergunta. Ele faz isso reduzindo cada função em um valor de acumulação que acontece ser um observable.

Lembre-se, o RxJS nos leva para fora

const chainRequestsWithProgress = (...requestFns) => (initialParams) =>  {
  const progress$ = new Rx.BehaviorSubject(0);
  const wrappedFns = requestFns.map((fn, i) => (...args) =>
    fn(...args).pipe(op.tap(() => progress$.next((i + 1) / requestFns.length)))
  );
  const chain$ = Rx.defer(() => {
    progress$.next(0);
    return chainRequests(...wrappedFns)(initialParams);
  });
  return [chain$, progress$];
};

chainRequestsWithProgress retorna dois observáveis - aquele que eventualmente emite a última resposta, e aquele que emite valores de progresso quando o primeiro observável é assinado. Fazemos isso criando um BehaviorSubject para servir como nosso fluxo de valores de progresso, e envolvendo cada uma de nossas funções de requisição para retornar o mesmo observável que normalmente retornariam, mas também encadeamos com tap para que possa enviar um novo valor de progresso para o BehaviorSubject.

O progresso é zerado a cada inscrição no primeiro observável.

Se você quisesse retornar um único observável que produzisse o estado de progresso, bem como o valor de resultado eventual, você poderia fazer com que chainRequestsWithProgress retornasse em vez disso:

chain$.pipe(
  op.startWith(null),
  op.combineLatest(progress$, (result, progress) => ({ result, progress }))
)

e você terá um observable que emite um objeto representando o progresso em direção ao resultado final, e depois esse resultado em si. Algo para refletir - será que progress$ precisa emitir apenas números?

Aviso

Isso assume que os observáveis de solicitação emitem exatamente um valor.

Resposta #2

A solução mais simples seria ter uma variável de contador de progresso que é atualizada a partir de um toque quando cada resposta retorna.

    let progressCounter = 0;

    request1.pipe(
      tap(_ => progressCounter = 0.33),
      concatMap(n => request2(n).pipe(
        tap(_ => progressCounter = 0.66),
        concatMap(n => request3(n)
          .pipe(tap(_ => progressCounter = 1)))
      ))
    );

Se você quer que o próprio progresso seja observável, então você quer compartilhar os observáveis da solicitação para não fazer solicitações duplicadas e, em seguida, combiná-los para obter o progresso.

Um exemplo de como você pode querer abordar isso pode ser encontrado em: https://www.learnrxjs.io/recipes/progressbar.html