这个 RxJs pipe() 链理解正确吗？ [关闭]

Question

我正在从事一个涉及 RxJS 的 Angular 项目，我对其他人编写的这段代码的工作原理有些怀疑（它工作正常）。我对 RxJS 不是很感兴趣，我正在为此苦苦挣扎。

基本上此代码用于执行搜索。在前端有一个搜索表单字段，用户可以在其中插入要搜索到产品集合中的单词。

所以在我的 HTML 页面中，我有这样的内容：

<input class="mx-2" type="search" placeholder="Cerca qui" [formControl]="filter">

然后在我的 TypeScript 代码中我有这个定义：

filter = new FormControl('');

由于用户会将字符一个一个地插入到先前的输入表单中，它会以这种方式监听 ngOnInit() 以了解 filter 变量值的变化：

this.filter.valueChanges
.pipe(
  startWith(''),
  debounceTime(300),
  takeUntil(this.destroy$),
  map((t) => (typeof t === 'number' ? t.toString() : t)),
  tap((text) =>
    !!wineList
      ? this.listWines$.next(this.search(text, wineList))
      : null
  )
)
.subscribe();

在这里我发现了一些困难，试图理解前面的代码部分是如何工作的。这是我的解释，如果我做错了断言，请纠正我：

filter FormControl 的 valueChanges 属性是一个 Observable，每当控件发生变化时（基本上是当用户插入一个新的字符或从搜索字符串中删除一个字符）。因此，订阅此事件是为了检测更改。因此，每次在输入表单中添加\删除新字符时，定义到 RxJS pipe() 运算符中的操作都会一个一个地执行（作为一个操作链）。

这里是这条链的行为：

首先startWith('')在新搜索到的字符前面添加一个空格。它自己发出一个新的 Observable 对象，表示这个“新”插入的字符。这个新的 Observable 将被链的下一步处理。

链的下一个步骤是 debounceTime(300)，它仅用于在 300 毫秒后发出一个 Observable，重新发送插入的字符串值（朝向下一个链步骤）。我想这样做是为了避免在多个用户进行任何搜索期间性能下降。

然后链中的下一步是 takeUntil(this.destroy$) 在此 this.destroy$ 可观察对象之后发出事件（代表我搜索的字符串）发射（我不知道为什么......但我认为它并不那么重要）

他们在链条的下一步是关键点：

map((t) => (typeof t === 'number' ? t.toString() : t)),

基本上 ma​​p() RxJS 运算符采用一个函数来更改我的 String 的值并发出一个表示新字符的新 Observable 对象。在这种情况下，如果 char 是一个数字，它将被转换为一个字符串。

最后使用 tap() 操作符来执行副作用，在这种情况下，副作用是每次将新字符添加到搜索中时执行新的 search() 操作表单域。它本身会发出一个新的 Observable 对象，其中包含检索到的产品列表。

我的推理正确吗？

Answer 1

this.filter.valueChanges.pipe(

  // start the stream by emitting an empty string.
  // this is only done once - not per emitted value.
  // the reasoning behind this, is that your pipeline expects strings or such
  // and to avoid errors, its initialized with an empty string as its first emit.
  startWith(''),

  // debounce the values that are being emitted.
  // this is used to prevent doing some expensive logic, for every key press (value change)
  // so this actually defers emissions by 300 milliseconds BEFORE actually "emitting" the last value.
  debounceTime(300),
  
  // every time you subscribe, you need to unsubscribe.
  // in order to avoid unsubscribing each and every subscription
  // we usually create a local subject, name it "destory" or something,
  // `destroy = new Subject<any>()`
  // and have it emit a value when component destroys
  // `ngOnDestroy() { this.destroy.next(); }`
  // then use the takeUntil operator on every stream to unsubscribe itself when the destroy subject emits
  takeUntil(this.destroy$),

  // if user enters a number, cast to string, else, keep it as is
  map((t) => (typeof t === 'number' ? t.toString() : t)),

  // tap is similar to placing the logic inside the "next" block.
  // it's usually used to either have side-effects, or, when you wish not to subscribe manually
  // which is why here, this is actually quite redundant and belongs to the .subscribe callback which is empty for some reason.
  tap((text) =>
    !!wineList
      ? this.listWines$.next(this.search(text, wineList))
      : null
  )
)
.subscribe();