RxJs неопределенные наблюдаемые вызовы

Question

Мне нужно получить неопределенный объем данных с сервера.Это должно идти следующим образом:

1. отправить первоначальный запрос
2. получить часть данных и сообщить серверу, что все в порядке, и я могу получить еще немного данных
3. повторяйте шаги 2 и 3 до тех пор, пока я не получу конкретное значение, означающее, что данных больше нет

Как я могу сделать это с помощью наблюдаемых?

Пока я могу думать только о рекурсивных наблюдаемых вызовах без использования какой-либо функции, например.

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();
        }
    });
}();

Answer 1

Не проверено, но вы можете попробовать этот шаблон

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

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

Answer 2

Следующее должно повторять вашу идею: 1 событие, выпущенное до завершения.

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);

Вы можете видеть, как оно работает в этом блице

Answer 3

как то так:

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