Используйте RxJs Pipe, чтобы уменьшить Observable к другому типу

Question

У меня есть Observable<Recipe[]>, который я хочу сократить до массива другого класса ChartData[], чтобы использовать его в качестве источника данных для графа старших диаграмм (столбец и круговая диаграмма).

Я пытаюсь использовать оператор канала RxJS на Observable<Recipe[]>, чтобы вызвать оператор сокращения для моих данных, но я не могу заставить его работать? Оператор reduce не выполняет итерации по этим элементам в моем Observable<Recipe[]> Ниже приведена моя попытка:

this.foodService.getAllReceipes()
  .pipe(
    reduce((array: ChartData[], value: Recipe[], i: number) => {
        const author = this.createOrFindAuthor(array, value[i]);
        author.y += 1;

        return array;
      }, new Array<ChartData>())
  )
  .subscribe(data => this.chartData$ = of(data.sort((a, b) => b.y - a.y)));
}

getAllRecipes() возвращает Observable<Recipe[]>

this.chartData$ является Observable<ChartData[]>

Я пытаюсь уменьшить это до ChartData[]. Я смог сделать это в операторе subscribe, и на графиках отображаются ожидаемые данные, но я подумал, что смогу сделать это как оператор с возможностью передачи данных? Вот что делается в рамках подписки:

this.foodService.getAllReceipes()
  .subscribe((data) => {
    const list = data.reduce((arr: ChartData[], v: Recipe) => {
      const author = this.createOrFindAuthor(arr, v);
      author.y += 1;

      return arr;
    }, new Array<ChartData>());

    this.chartData$ = of(list.sort((a, b) => b.y - a.y));
  });

Я пытался использовать код subscribe в конвейере reduce, но я получаю ошибки компиляции, говорящие, что метод ожидает Recipe[] для значения. Но если я использую массив, тогда я получу только первый элемент из Observable (или я просто получаю Observable и мне нужно что-то с этим сделать?)

Возможно ли это или мой мыслительный процесс неверен в отношении того, как должен работать конвейерный оператор в Observable?

Для справки вот модели и функция createOrFindAuthor:

export class Recipe {

    public Title: string;
    public Author: string;
    public Source: string;
    public Page: number;
    public Link?: string;
}

export class ChartData {
    name: string;
    y: number;
}

private createOrFindAuthor(array: ChartData[], recipe: Recipe): ChartData {
  const name = (recipe.Author.length > 0 ? recipe.Author : 'UNKNOWN');

  let found = array.find(i => i.name === name);

  if (!found) {
    const newData = new ChartData();
    newData.name = name;
    newData.y = 0;
    array.push(newData);

    found = newData;
  }

  return found;
}

Answer 1

Итак, Чау Тран поставил меня на правильные позиции. Очевидно, мне нужно было switchMap Observable для Recipe[], и оператор reduce тогда был рад принять Recipe в качестве значения. Решение следующим образом:

this.foodService.getAllReceipes()
  .pipe(
    switchMap(data => data as Recipe[]),            <<== ADDED THIS

    reduce((array: ChartData[], value: Recipe) => {
        const author = this.createOrFindAuthor(array, value);
        author.y += 1;

        return array;
      }, new Array<ChartData>()),

      switchMap(data => this.chartData$ = of(data.sort((a, b) => b.y - a.y)))
  )
  .subscribe();

Answer 2

Я создал этот пример stackblitz для демонстрационного использования reduce(). В проекте есть другие вещи, но вам нужно: demoReduce.ts:

import { Observable, of } from 'rxjs'
import { reduce, tap } from 'rxjs/operators'

type Book = {
  title: string
  noPages: number
}

type Library = {
  totalPages: number
  books: Book[]
}

export const demoReduce = () => {
  const books$: Observable<Book> = of(
    { title: 'book 1', noPages: 10 },
    { title: 'book 2', noPages: 20 },
    { title: 'book 3', noPages: 30 },
  )

  return books$.pipe(
    // --- reduce a stream of "Book" into a "Library"
    reduce<Book, Library>((previous, book) => {
      // --- add book to "Library" and increment totalPages in "Library"
      return {
        totalPages: previous.totalPages + book.noPages,
        books: [
          ...previous.books,
          book
        ]
      }
    }, { totalPages: 0, books: [] }),
    tap(val => console.log(val))
  )
}

Чтобы выполнить наблюдаемое, используйте кнопку «Demo Reduce» и посмотрите на консоль.

Преобразует поток Books в один Library объект.

Примечание:

• Я не уверен, почему stackblitz показывает ошибку (волнистое красное подчеркивание) в онлайн-редакторе для reduce(). Я не получил ошибку в IntelliJ / WebStorm. Я подозреваю, что это ошибка стекаблика.

Обновление:

Вот та же функция, которая принимает Observable<Book[]> в качестве ввода (не проверено):

export const demoReduceWithArray = () => {
  const books$: Observable<Book[]> = of([
    { title: 'book 1', noPages: 10 },
    { title: 'book 2', noPages: 20 },
    { title: 'book 3', noPages: 30 }
  ])

  return books$.pipe(
    // --- reduce a stream of "Book[]" into a "Library"
    reduce<Book[], Library>((previous, books) => {
      // --- add each book to "Library" and increment totalPages in "Library"
      books.map(book => {
        previous = {
          totalPages: previous.totalPages + book.noPages,
          books: [
            ...previous.books,
            book
          ]
        }
      })
      return previous
    }, { totalPages: 0, books: [] }),
    tap(val => console.log(val))
  )
}

Answer 3

После reduce попробуйте:

switchMap(data => {
    this.chartData$ = of(data.sort((a, b) => b.y - a.y));
    return this.chartData$;
})
.subscribe()