1
votes

Appels observables indéfinis RxJs

J'ai besoin de récupérer une quantité indéfinie de données sur le serveur. Cela devrait se dérouler de la manière suivante:

envoyer la demande initiale
récupérer une partie des données et dire au serveur que tout va bien et je peux obtenir plus de données
répétez les étapes 2 et 3 jusqu'à ce que je reçoive une valeur spécifique qui signifie qu'il n'y a plus de données

Comment puis-je faire cela en utilisant des observables?

Pour l'instant, je ne peux penser qu'aux appels observables récursifs à partir d'une fonction comme.

const send = execSend() {
    this.send(message).subscribe(resp => {
        if (resp === 'end') {
            subscriber.next(byteArr.join(''));
            console.log('finished');
            subscriber.complete();
        } else {
            byteArr.push(resp);
            execSend();
        }
    });
}();

typescript angular rxjs angular2-observables

2 commentaires

Voulez-vous envoyer la même demande à chaque fois?

Oui. C'est juste pour dire au serveur que j'ai ma valeur et que je suis prêt pour le prochain. Pour être plus précis, tout cela consiste à récupérer une énorme donnée par morceaux pour tout combiner lorsque tout est reçu. Puisqu'un WebSocket est utilisé, nous sommes limités aux limites de taille des données.

3 Réponses :

-1
votes

quelque chose comme ceci:

let todo = true;
interval(100).pipe(
    takeWhile(()=>todo),
    concatMap(()=>getStuff())
).subscribe(data => {
  todo = !isFinished(data);
});

0 commentaires

0
votes

Non testé mais vous pouvez essayer ce modèle

exec=()=>http.get(....) 

exec().pipe(
  expand((resp)=>exec()),
  takeWhile(resp=>resp !== 'end'),
  scan((acc,curr)=>acc.concat(curr),[])
).subscribe()

0 commentaires

1
votes

Ce qui suit devrait reproduire votre idée: 1 événement émis avant la fin.

import { of } from 'rxjs';
import { expand, takeWhile, reduce } from 'rxjs/operators';

let count = 0;
const FINISH = "finished";
const limit = 5;
const send$ = () => of(count++ < limit ? "sent" : FINISH);

const expander$ = send$().pipe(
  expand(resp => send$()),
  takeWhile(resp => resp !== FINISH),
  reduce((acc, val) => acc ? acc + val : val, null)
);

const subscribe = expander$.subscribe(console.log);

Vous pouvez le voir fonctionner dans ce blitz

12 commentaires

Cela ne plantera-t-il pas si mon observable (envoi) se termine juste après avoir passé une valeur?

Pourriez-vous être plus précis?

Mon observable émet de la valeur et termine l'abonnement juste après cela.

L'exemple fait de même, send $ crée une observable qui émet une valeur et se termine ...

Merci je vais essayer. Pourquoi send $ est-il une fonction? Pas seulement send $ = of (...)

Bonne question! Deux raisons: simuler la création d'une nouvelle requête et s'assurer que le compteur monte. Si vous refactorisez envoyer $ pour être une référence, l'échantillon débordera car l'expression count ++ sera évaluée lors de l'appel de of , une fois par exemple. Vous m'avez pris au dépourvu là-bas, m'avez pris quelques minutes pour réaliser les conséquences que cela pouvait avoir.

@Sergey envisage d'accepter cette réponse si elle a résolu votre problème, afin que votre question puisse être préservée.

Je sais. Je n'ai rien essayé pour dire que cela a résolu mon problème. J'ai besoin de temps.

@Sergey une chance?

Pas encore. Avoir du travail à faire avant de le tester (besoin d'une API d'abord)

Finalement, je lui ai donné une chance. Fonctionne bien sauf que maintenant, je dois ne pas répondre au dernier message. Désolé pour une telle latence.

Avec la modification expand (resp => resp! == 'finish'? This.send ('ok'): empty ()), fonctionne comme prévu