在 RxJS 中重用 pipeable 操作符

Question

我在一个 Angular 组件中有两个主题，它们利用同一组可管道运算符为两个不同的表单字段提供预先输入的搜索查找。示例：

this.codeSearchResults$ = this.codeInput$
                               .pipe(
                                 untilDestroyed(this),
                                 distinctUntilChanged(),
                                 debounceTime(250),
                                 filter(value => value !== null),
                                 switchMap((value: string) => {
                                   const params: IUMLSConceptSearchParams = {
                                     ...TERMINOLOGY_SEARCH_PARAMS,
                                     sabs: this.sabs,
                                     term: value
                                   };

                                   return this.terminologyService.umlsConceptSearch(params);
                                 }),
                               );

definition for pipe 似乎可以接受任意数量的函数，但通过 spread 提供函数

this.codeSearchResults$ = this.codeInput$.pipe(...operators);

没有按预期工作。如何为两个 Subject 提供单一输入函数源以保持我的代码 DRY？

编辑

按照 Dan Kreiger 的回答中的选项 #2，我的最终代码如下：

const operations = (context) => pipe(
      untilDestroyed(context),
      distinctUntilChanged(),
      debounceTime(250),
      filter(value => value !== null),
      switchMap(value => {
        const term: string = value as unknown as string;
        const params: IUMLSConceptSearchParams = {
          ...TERMINOLOGY_SEARCH_PARAMS,
          sabs: context.sabs,
          term,
        };

        return context.terminologyService.umlsConceptSearch(params) as IUMLSResult[];
      }),
    );

    this.codeSearchResults$ = this.codeInput$
                                  .pipe(
                                    tap(() => this.codeLookupLoading = true),
                                    operations(this),
                                    tap(() => this.codeLookupLoading = false),
                                  ) as Observable<IUMLSResult[]>;

    this.displaySearchResults$ = this.displayInput$
                                      .pipe(
                                        tap(() => this.displayLookupLoading = true),
                                        operations(this),
                                        tap(() => this.displayLookupLoading = false),
                                      ) as Observable<IUMLSResult[]>;

我需要编写几个 tap() 函数，它们对每个主题都是唯一的，并且可以按预期工作。

Answer 1

这里有 4 种可能的方法。

1.返回运算符列表的实用函数

如果您想在上下文之间重用它，可以尝试创建一个接受thisArg 并返回运算符数组的函数。

然后你可以在传递给pipe的参数中传播调用的函数。

/**
 * @param {object} thisArg - context using the typeahead
 * @returns {OperatorFunction[]}
 * 
 * list of pipepable operators 
 * that can have a dynamic `this` context
 */
const typeAhead = thisArg => [
  untilDestroyed(thisArg),
  distinctUntilChanged(),
  debounceTime(250),
  filter(value => value !== null),
  switchMap((value: string) => {
    const params: IUMLSConceptSearchParams = {
      ...TERMINOLOGY_SEARCH_PARAMS,
      sabs: thisArg.sabs,
      term: value
    };

    return thisArg.terminologyService.umlsConceptSearch(params);
  })
]


// subject A
this.codeSearchResults$ = this.codeInput$
  .pipe(...typeAhead(this));

// subject B
this.articleSearchResults$ = this.articleInput$
  .pipe(...typeAhead(this));

注意： terminologyService 和 sabs 需要出现在您传递给此函数的 this 上下文中。

看起来您正在将这些用于组件，所以只要 terminologyService 作为依赖项被注入并且 sabs 是组件的静态成员，它应该可以工作。

---

2。返回组合运算符的实用函数

或者，您可以通过从 rxjs 导入 pipe 以将运算符链接在一起来实现相同的目的。

import { pipe } from "rxjs";

// ... 

const typeAhead = (thisArg) =>
  pipe(
    untilDestroyed(thisArg),
    distinctUntilChanged(),
    debounceTime(250),
    filter((value) => value !== null),
    switchMap((value: string) => {
      const params: IUMLSConceptSearchParams = {
        ...TERMINOLOGY_SEARCH_PARAMS,
        sabs: thisArg.sabs,
        term: value
      };

      return thisArg.terminologyService.umlsConceptSearch(params);
    })
  );

在这种情况下，您不需要使用扩展运算符，因为运算符是使用 pipe 的从左到右的函数组合组合在一起的。

// subject A
this.codeSearchResults$ = this.codeInput$
  .pipe(typeAhead(this));

// subject B
this.articleSearchResults$ = this.articleInput$
  .pipe(typeAhead(this));

注意： terminologyService 和 sabs 需要出现在您传递给此函数的 this 上下文中。

---

3.自定义运营商服务

您可以为自定义管道运算符创建可重用服务。这将允许您从可重用服务本身中获取 terminologyService 单例。

但是，看起来sabs 仍然需要从任何来源获得。

如果您选择这样做，请确保在顶级提供程序中声明 TerminologyService - 请参阅示例 here

然后你可以将它注入到你的组件中。

import { TerminologyService } from "./terminologyService.service";
import { Injectable } from "@angular/core";
import { UntilDestroy, untilDestroyed } from '@ngneat/until-destroy';


@UntilDestroy()
@Injectable({
  providedIn: 'root',
})
export class PipedOperatorsService {
  constructor(private terminologyService: TerminologyService) {}

  typeAhead(thisComponent) {
    return pipe(
      untilDestroyed(thisComponent),
      distinctUntilChanged(),
      debounceTime(250),
      filter((value) => value !== null),
      switchMap((value: string) => {
        const params: IUMLSConceptSearchParams = {
          ...TERMINOLOGY_SEARCH_PARAMS,
          sabs: thisComponent.sabs,
          term: value
        };

        return this.terminologyService.umlsConceptSearch(params);
      })
    );
  }
}

然后你可以在组件中使用它：

import { Component, OnInit } from "@angular/core";
import { PipedOperatorsService } from "./pipedOperators.service";
import { Observable } from 'rxjs';

@Component({
  selector: "some-root",
  templateUrl: "./some.component.html",
})
export class SomeComponent implements OnInit {
  sabs = ['What', 'is', 'a', 'sab', '?', '?']; 
  codeSearchResults$: Observable<string[]>;

  constructor(private pipedOperatorsService: PipedOperatorsService){}

  ngOnInit() {
    this.codeSearchResults$ = this.codeInput$.pipe(
      this.pipedOperatorsService.typeAhead(this)
    );
  }
}

注意： sabs 需要出现在您传递给此函数的 this 上下文中。我在这里做了一个假人。

---

4.自定义运算符的服务（无 this 上下文）

如果您想让它真正可重用且不必担心this，您只需编写不需要this 上下文的运算符即可。

import { Injectable } from "@angular/core";
import { pipe } from "rxjs";
import { debounceTime, distinctUntilChanged, filter } from 'rxjs/operators'


@Injectable({
  providedIn: 'root',
})
export class PipedOperatorsService {    
  get typeAhead() {
    return pipe(
      distinctUntilChanged(),
      debounceTime(250),
      filter((value) => value !== null),
    );
  }
}

然后一定要在组件中添加你需要的具体操作符：

import { Component, OnInit } from "@angular/core";
import { TerminologyService } from "./terminologyService.service";
import { PipedOperatorsService } from "./pipedOperators.service";
import { UntilDestroy, untilDestroyed } from '@ngneat/until-destroy';
import { switchMap } from 'rxjs/operators';
import { Observable } from 'rxjs';


@UntilDestroy()
@Component({
  selector: "some-root",
  templateUrl: "./some.component.html",
})
export class SomeComponent implements OnInit {
  sabs = ['What', 'is', 'a', 'sab', '?', '?']; 
  codeSearchResults$: Observable<string[]>;

  constructor(private pipedOperatorsService: PipedOperatorsService, private terminologyService: TerminologyService){}

  ngOnInit() {
    this.codeSearchResults$ = this.codeInput$.pipe(
      untilDestroyed(this),
      this.pipedOperatorsService.typeAhead,
      switchMap((value: string) => {
        const params: IUMLSConceptSearchParams = {
          ...TERMINOLOGY_SEARCH_PARAMS,
          sabs: this.sabs,
          term: value
        };

        return this.terminologyService.umlsConceptSearch(params);
      })
    );
  }
}

注意： sabs 需要出现在您传递给此函数的 this 上下文中。我在这里做了一个假人。

我已经有一段时间没有使用 Angular 了。我希望这些示例中的一些有用。