Интервал RxJS, запрос AJAX и уменьшение

Question

Я пытаюсь выяснить, как отправлять запросы AJAX каждые пять секунд, отображая и сокращая данные с помощью операторов RxJS.

Итак, существует простой сервис Angular, который запрашивает внешний API для астронавтов, которые в данный момент находятся в космосе: http://api.open -notify.org / astros.json

В службе я сначала пытаюсь установить interval(5000), а затем с помощью оператора сопоставления RxJS сопоставить входящий номер с запросом GET.

Задача заключается в следующем: мне нужно сначала найти астронавтов, затем найти тех, кто летит на МКС, а затем тех астронавтов, которых я могу визуализировать. Данные должны обновляться каждые 5 секунд; поэтому мне нужно отправлять один и тот же HTTP-запрос каждые 5 секунд, что я могу сделать с setInterval(). И это не самое чистое решение проблемы.

Вот код:

import { Injectable } from '@angular/core';
import { HttpClient, HttpErrorResponse } from '@angular/common/http';
import { Observable, throwError, interval } from 'rxjs';
import { catchError, filter, concatMap, flatMap, map, reduce } from 'rxjs/operators';
import { Astronaut } from '../models/astronaut.model';

const astronautsUri = 'http://api.open-notify.org/astros.json';

@Injectable({
  providedIn: 'root'
})
export class IssLocatorService {
  constructor(private http: HttpClient) {}

  getAstronauts(): Observable<Astronaut[]> {
    return interval(5000).pipe(
      concatMap(num => this.http.get(astronautsUri)),
      flatMap((newAstronauts: any) => newAstronauts.people),
      filter((astronaut: Astronaut) => astronaut.craft === 'ISS'),
      map((astronaut: Astronaut) => [astronaut]),
      reduce((prev, next) => [...prev, ...next]),
      catchError(this.handleError)
    );
  }
}

Код не работает, увы. Хотя поток доходит до оператора reduce(), оператор не возвращается к методу subscribe() в компоненте.

Однако мне кажется, что решение должно работать нормально. Вот как я думаю, что это работает:

1. Через 5 секунд создается первое число 0, и внешняя наблюдаемая отображается на внутреннюю наблюдаемую - запрос AJAX. concatMap() ждет, пока завершится внутренняя Наблюдаемая, и только после этого получает второе число - 1.
2. Внутренний Observable - это HTTP-запрос, который возвращает JSON. Мне нужны только люди из объекта, который выглядит примерно так: {success:true, people: [...]}), поэтому я использую flatMap() для преобразования этого объекта в массив астронавтов people. Каждый объект в people становится Наблюдаемым благодаря тому, как работает flatMap().
3. Я фильтрую каждый объект астронавта.
4. Я сопоставляю каждый объект астронавта с уменьшенным массивом.
5. Я уменьшаю массив космонавтов. Другими словами, astronauts.people воспроизводится благодаря reduce().
6. (Вот в чем проблема) reduce(), в соответствии со спецификацией, должно вернуться к subscribe, потому что внутренняя наблюдаемая завершена. Но он не : reduce() ждет, пока следующее число не будет сгенерировано interval() и сопоставлено с внутренней наблюдаемой, снова people возвращаются и помещаются в тот же массив. И это продолжается и продолжается.

Если я заменим reduce() на scan, массив или астронавты вернутся к методу subscribe. Однако этот массив постоянно увеличивается из-за того, что в него постоянно помещают объекты астронавтов.

Следующий подход работает просто отлично:

return this.http.get(astronautsUri).pipe(
  flatMap((newAstronauts: any) => newAstronauts.people),
  filter((astronaut: Astronaut) => astronaut.craft === 'ISS'),
  map((astronaut: Astronaut) => [astronaut]),
  reduce((prev, next) => [...prev, ...next]),
  catchError(this.handleError)
);

Но в этом случае мне нужно вручную установить интервал с setInterval() в классе компонентов, который отображает астронавтов, и мне нужно вызвать метод getAstronauts(). Итак, в ngOnInit есть два вызова метода, в основном.

Как мне достичь желаемого эффекта только с помощью операторов RxJS? Я хочу установить интервал, отобразить, отфильтровать и уменьшить массив объектов, а затем получить их.

Мое понимание того, как работает отображение RxJS, действительно плохое, но я попробовал (ради попытки) все эти методы - switchMap(), concatMap(), exhaustMap(), flatMap() - чтобы отобразить числа из interval() к запросу AJAX. Это все еще не работает.

Answer 1

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

const { from } = rxjs;
const { scan } = rxjs.operators;

const obs$ = from([1,2,3,4,5]);

obs$.pipe(scan((total, item) => total + item)).subscribe(val => { console.log(val); });

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.4.0/rxjs.umd.min.js"></script>

const { from } = rxjs;
const { reduce } = rxjs.operators;

const obs$ = from([1,2,3,4,5]);

obs$.pipe(reduce((total, item) => total + item)).subscribe(val => { console.log(val); });

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.4.0/rxjs.umd.min.js"></script>

Answer 2

Я думаю, что вы пытаетесь достичь

getAstronauts(): Observable<Astronaut[]> {
    return interval(5000).pipe(
      concatMap(num => 
        this.http.get(astronautsUri).pipe(
          flatMap((newAstronauts: any) => newAstronauts.people),
          filter((astronaut: Astronaut) => astronaut.craft === 'ISS'),
          reduce((prev, astronaut) => [...prev, astronaut], []),
        )
      ),
      catchError(this.handleError)
    );
  }

Проблема метода Reduce состоит в том, что он ожидает завершения своего источника, наблюдаемого, прежде чем выдавать какое-либо значение. в моем коде источник содержит элементы одного запроса, а в примере с вопросом источник - это все элементы всех запросов, которые никогда не заканчиваются