rxjs nova tentativa com atraso e multicast

Question

Estou tentando construir uma cadeia rxjs que, em última análise, chama uma função de biblioteca que retorna uma promessa (ou seja, não posso alterar nada além dessa chamada):

  libraryCall(): Promise<SomeOutput> { /* opaque to me */ }

A referida libraryCallpromessa pode ser resolvida ou rejeitada (como qualquer promessa).

Se a promessa for resolvida, então tudo deve funcionar como uma chamada simples, nada mais adicionado.

Se a promessa for rejeitada, deve-se tentar novamente (ligar libraryCallnovamente) indefinidamente (sim, indefinidamente, já que essa promessa retorna uma informação muito importante), após algum tempo constante (1s por exemplo).

O observável resultante será emitido apenas uma vez (depois da libraryCallrejeição da promessa de N e da libraryCallresolução da promessa de 1 ).

Se algum novo cliente assinar o observável resultante enquanto ele está em loop, tentando novamente para sempre, ele deve pegar carona nesse loop. Assim que uma promessa for resolvida, todos os clientes pendentes devem obter esse valor resolvido.

Como último requisito, se cada cliente pendente cancelar a assinatura, o loop de repetição deverá ser cancelado.

O que eu fiz até agora?

    defer(() => libraryCall()).pipe(
      retryWhen(errors => errors.pipe(delay(1000))),
      shareReplay(0),
    );

Está quase funcionando. Mas ... depois de resolver uma vez, qualquer nova assinatura emite imediatamente o último valor resolvido. Preciso ligar libraryCallnovamente nesses casos.

Tenho certeza de que tem algo a ver com isso shareReplay(0), embora eu tenha definido seu buffer para 0esperar que nada seja armazenado em buffer (ou seja, preciso da parte "compartilhar", não da parte "repetir" shareReplay).

Como posso corrigir (ou reescrever)?

Answer 1

Mas ... depois de resolver uma vez, qualquer nova assinatura emite imediatamente o último valor resolvido

Isso acontece possivelmente por causa de como o ReplaySubjecttrata o caso quando o bufferSizeargumento é0 :

/* ... */

this.bufferSize = Math.max(1, bufferSize);

/* ... */

Portanto, embora você esteja passando 0como bufferSize, ele será definido bufferSizecomo 1, o que deve explicar por que, quando um novo assinante se registra, ele obterá o valor armazenado em buffer, em vez de invocar a função de biblioteca novamente.

Acho que uma maneira rápida de resolver isso seria:

defer(() => libraryCall()).pipe(
  retryWhen(errors => errors.pipe(delay(1000))),
  shareReplay({
    bufferSize: 1,
    refCount: true,
  }),
  first(),
);

Ao usar a refCount: trueopção, ele garante que quando não houver mais assinantes ativos, ele fará uma nova assinatura da fonte observável, recuperando assim a função, quando um novo assinante se registrar.

Acho que ajudaria ver o que acontece no código-fonte.

Quando todo o stream estiver inscrito :

if (!subject) {
  // subscribed for the first time

  subject = new ReplaySubject<T>(bufferSize, windowTime, scheduler);

  // adding the new subscriber to the `ReplaySubject`'s subscribers list
  // (the `ReplaySubject` extends `Subject`, so this is why it also has a list of subscribers)
  innerSub = subject.subscribe(subscriber);

  // subscribing to the source - this will cause the library function to be called
  subscription = source.subscribe({
    next(value) { subject!.next(value); },
    error(err) {
      const dest = subject;
      subscription = undefined;
      subject = undefined;
      dest!.error(err);
    },
    complete() {
      subscription = undefined;
      subject!.complete();
    },
  });

  // The following condition is needed because source can complete synchronously
  // upon subscription. When that happens `subscription` is first set to `undefined`
  // and right after is set to the "closed subscription" returned by `subscribe`
  if (subscription.closed) {
    subscription = undefined;
  }
} else {
  // subscribed for the second, third etc... time

  // when other subscribers register, they will be added to the `ReplaySubject`'s subscribers list
  // so that every time the source emits, each subscriber will get the same value
  innerSub = subject.subscribe(subscriber);
}

Ao usar first(), após um assinante receber um valor, um completeevento ocorrerá, significando que o assinante será removido da lista de assinantes do assunto.

Para a shareReplayoperadora, é o que acontece quando um assinante é retirado da lista (devido a uma complete/ errornotificação):

subscriber.add(() => {
  refCount--;
  innerSub.unsubscribe();
  if (useRefCount && refCount === 0 && subscription) {
    subscription.unsubscribe();
    subscription = undefined;
    subject = undefined;
  }
});

Como você pode ver, devido a refCount === 0(não há mais assinantes na lista) e useRefCount( refCount: true), você deve obter os resultados esperados. Se if blockfor atingido, subjectse tornará undefined, o que significa que quando um novo assinante se inscrever no stream, ele alcançará o if (!subject) { ... }bloco, portanto, a fonte será assinada novamente.

Além disso, eu diria que este requisito também será atendido:

se cada cliente pendente cancelar a assinatura, o loop de repetição deverá ser abortado.

---

Em uma nota lateral, desde que você não precisa da repetição parte, eu acho que você poderia substituir shareReplay()com share(). ( EDITAR: notei o comentário tarde demais ).