RxJS 6有条件地管道一个可观察的

Question

我有一个现有的函数,它执行一些调用HTTP端点的工作,负责管理我的Angular组件中的一些逻辑.需求的来源是,在某种情况下,我们需要调用所有现有逻辑,还要进行另一个HTTP调用.以下是我的尝试的注释版本.

public resolveComponent(action: string) {
        //if called with no parameters, don't do anything
        //otherwise, call one of the two endpoints.     
        let preAction = (x: any) => {
            if (action === "replay") {
                return this.remediationService
                           .replayRemediation(this.workflow.Id);
            }
            else if (action === "remove") {
                return this.remediationService
                           .removeRemediation(this.workflow.Id);
            }
            else {
                return of([]);
            }
        }

        this.remediationService
            .resolveComponent(component)
            .pipe(
                //the line below was what I added after new requirement
                mergeMap(preAction),
                mergeMap(x => {
                    //grabs updated data from server
                    return this.remediationService.remediationComponent(component.Id);
                })
            )
            .subscribe((x: SomeModel) => {
                    //logic happens
                });
}

这按预期执行工作,但这是有条件地链接可观察量的最佳方式吗？

Answer 1

使用iifrxjs。

iif接受一个条件函数和两个 Observable。* 当订阅了操作符返回的 Observable 时，将调用条件函数。

https://www.learnrxjs.io/operators/conditional/iif.html

Answer 2

iif可以有条件地写入可观察量：

• iif(() => cond, obs1, obs2)=cond ? obs1 : obs2
• iif(() => cond, obs1)=cond ? obs1 : EMPTY

但iif不适用于运营商。如果您想有条件地在 中写入运算符.pipe()，请使用：

cond ? delay(1000) : interval(500)

对于空运算符，请使用identity：

import { identity } from 'rxjs'
cond ? delay(1000) : identity

identity实际上在哪里x => x

Answer 3

您可以添加自己的pipeIf：

import { from, Observable, of, NEVER } from 'rxjs';
import { mergeMap, map, filter } from 'rxjs/operators';

from([10, 11, 12, 13])
  .pipe(
    map(x => x + 10),
    pipeIf(x => x > 21,
      filter(x => x > 22),
      map(x => x + 100)
    ),
)
  .subscribe(console.log);

function pipeIf(predicate, ...pipes) {
  return function (source) {
    return source.pipe(
      mergeMap(value => predicate(value) ? of(value).pipe(...pipes) : of(value))
    )
  }
}

// 20
// 21
// 123