Угловой бесконечный цикл опроса HTTP без Rx.interval

Question

Я работаю с API, который реализует длинный цикл опроса, так как сервер сохраняет соединения открытыми в течение произвольного промежутка времени (скажем, от 0 до 5 с), а затем возвращает ответ, как только у него появляется новое сообщение дляпередача.

В настоящее время мой код отправляет запрос каждые 5 секунд, независимо от того, когда сервер отвечает.Если в строке 3 запроса и сервер отвечает в виде [1 с, 2 с, 3 с], то в настоящее время я буду отправлять 3 запроса [5 с, 5 с, 5 с] в общей сложности ~ 15 с, в идеале я бы хотел, чтобы всепроизойдет за 6 секунд (1 + 2 + 3).

Rx.interval(5000)
  .pipe(
    Rx.concatMap(() => httpClient.get('/api/messages')),
    retry(8000)
   )
  .subscribe((data) => handleResponse(data));

Если бы это было обещание, я бы написал что-то вроде

const fetchRequest = httpClient.get('/api/messages').toPromise()
  .then(data => {handleResponse(data); fetchRequest()});

, но в моем случае мне нужно было бы вернутьсяObvservable.

Из чтения документов RxJS, что, похоже, ближе всего к тому, что я хочу, будет retryWhen, который будет работать, но кажется мне семантически неправильным, так как заставит меня поднять Error для поддержания цикла.

httpClient.get('/api/messages')),
  .subscribe((data) => {
    handleResponse(data);
    throw 'keep going';
  })
  .retryWhen(val => val === 'keep going')

Есть ли более элегантный способ, чем использование retryWhen для обработки этого случая?

Answer 1

Насколько я понимаю, вы хотите отправить запрос на сервер, и через некоторое время, когда вы получите ответ, вы снова хотите отправить новый запрос.

Демонстрационная версия StackBlitz для этого кода

Это пример именно для этого поведения.

import { Observable, Subject, of } from 'rxjs';
import { switchMap, delay, startWith, tap, concatMap } from 'rxjs/operators';

@Component({
  selector: 'my-app',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css']
})
export class AppComponent {

  subject = new Subject();

  constructor() {

    this.subject.pipe(
      startWith({}),
      concatMap(() => this.testGet()),
      tap(() => this.subject.next()),
    ).subscribe(x => {
      console.log('inside subscribe', x);
    })

  }

  async testGet() {
    console.log('start get');
    await of({}).pipe(delay(2000)).toPromise();
    console.log('finish get');
    return null;
  }
}