使用 RxJs Pipe 将 Observable 减少为不同的类型

Question

我有一个 Observable<Recipe[]>，我想将其缩减为不同 class ChartData[] 的数组，以用作 highcharts 图形（柱形图和饼图）的数据源。

我正在尝试使用 Observable<Recipe[]> 上的 RxJS 管道运算符来调用我的数据的 reduce 运算符，但我无法让它工作？ 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() returns 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));
  });

我尝试在可管道 reduce 中使用 subscribe 代码，但我收到编译错误，指出该方法需要 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

在reduce之后，试试：

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

Answer 2

所以 Chau Tran 把我放在了正确的位置上。显然我需要 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 3

我创建这个 stackblitz example 是为了演示 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))
  )
}

要执行 observable，请使用 "Demo Reduce" 按钮，然后查看控制台。

它将 Books 的流转换为单个 Library 对象。

注：

我不确定为什么 stackblitz 在 reduce() 的在线编辑器中显示错误（红色波浪下划线）。我没有在 IntelliJ/WebStorm 中得到错误。我怀疑是一个 stackblitz 错误。

更新：

这是将 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))
  )
}

使用 RxJs Pipe 将 Observable 减少为不同的类型

Use RxJs Pipe to reduce Observable to different type

angular

typescript

rxjs

rxjs-pipeable-operators