RXJS Observable 移除管道操作符

Question

我有一个问题，我想删除/添加任何以前添加的 observable 运算符。

我得到了以下我无法更改的可观察值：

let objects$ = of([{
      category: 1,
      name: 'Some1'
    }, {
      category: 2,
      name: 'Some2'
    }]).pipe(
      map(o => o.filter(b => b.category === 2))
    )
    objects$.subscribe(obj => console.log(obj));

如预期的那样输出：{category: 2,name: 'Some2'}。

现在我想把过滤器的值改成b.category === 1输出{category: 1,name: 'Some1'}。

如果我执行以下操作：

objects$.pipe(
      map(o => o.filter(b => b.category === 1))
    )
    objects$.subscribe(obj => console.log(obj));

我仍然收到{category: 2,name: 'Some2'}。

如果我这样做：

objects$ = objects$.pipe(
      map(o => o.filter(b => b.category === 1))
    )
    objects$.subscribe(obj => console.log(obj));

我得到 [] 因为 observable 的输出不再具有类别 1。

我的问题是如何从原始 observable 中删除 .pipe() 以添加新的？

Answer 1

您不能删除运算符

通过使用 pipe 例如调用observable$.pipe( map(..) ) 你并没有真正在observable$ 中添加一些以后可以删除的东西，即使add operator X to your observable 这个短语经常被使用。

可管道操作符是一个将 Observable 作为其输入的函数 并返回另一个 Observable。这是一个纯操作：上一个 Observable 保持不变

const o1$ = of('1');
const o2$ = o1$.pipe(
  map(x => x + '2'),
  map(x => x + '3')
);

等价于

const o1$ = of('1');
const o2$ = map(x => x + '3')(
  map(x => x + '2')(
    o1$
  )
);

你基本上是用你原来的 Observable o1$ 作为输入和其他一些 Observable 作为输出来链接纯函数调用。 o1$ 不会更改，因此您必须将返回值分配给变量 (o2$) 以供以后使用或直接使用它才能产生任何效果。

在上面的示例中，您可以重用o1$ 并通过在o1$ 上使用不同的运算符来创建与o2$ 不同的Observable。但是，您不能从 o2$ 中删除用于定义 o2$ 的函数调用。

您应该能够更改您关心的实施细节

如果你得到了一个 Observable，你就不能改变你不应该关心它的具体实现，而只关心它发出的内容。在这种情况下，Observable 对您来说是一个黑盒子，据您所知，它只会发出 { category: 2, name: 'Some2' }。你不知道也不应该关心 Observable 在发出这个值的过程中所做的具体事情。

但是你似乎知道并关心 Observable 在发出这个值之前做了什么。如果你这样做了，你（或同事）必须有一种方法来改变 Observable 的实现。

动态注入代码

如果您的代码的一部分应该是固定的，而另一部分应该是动态的，您可以定义一个接受另一个函数（动态部分）作为输入的高阶函数（固定部分），并以这种方式从一个原始 Observable 创建不同的 Observable ：

let getObjects$ = (predicate: (value: any) => boolean) => of([
  {
    category: 1,
    name: "Some1"
  },
  {
    category: 2,
    name: "Some2"
  }
]).pipe(map(o => o.filter(predicate)));

getObjects$(v => v.category === 1).subscribe(obj => console.log(obj));
getObjects$(v => v.category === 2).subscribe(obj => console.log(obj));

Answer 2

您在第一个操作中应用了管道运算符 (objects$ => Observable.pipe())。 第一次操作后，objects$ 的内部值为：

[{category: 2,name: 'Some2'}]

然后您的第二次操作将在第一次的结果上完成（Observable.pipe(...).pipe(...)）。它不会改变objects$ 的值。 但是您再次订阅了最初的 observable (objects$ => Observable.pipe())。 第二个管道将不会改变值，而是在应用pipe 运算符函数后返回new Observable。

为了想要的结果：

let objects$ = of([
  {
    category: 1,
    name: "Some1"
  },
  {
    category: 2,
    name: "Some2"
  }
]);

objects$
  .pipe(map(o => o.filter(b => b.category === 1)))
  .subscribe(obj => console.log(obj));

objects$
  .pipe(map(o => o.filter(b => b.category === 2)))
  .subscribe(obj => console.log(obj));