在 RxJS 中，如果源 observable 没有发出任何值，如何防止操作符发出值？

Question

我有三个 observable foo$、bar$ 和 baz$，我将它们合并在一起形成另一个 observable。这按预期工作：

• 流以>>> 开头
• 每个值一个一个地发出
• 流以<<< 结尾

const foo$ = of('foo');
const bar$ = of('bar');
const baz$ = of('baz');

merge(foo$, bar$, baz$).pipe(startWith('>>>'), endWith('<<<')).subscribe(str => {
  console.log(str);
});

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.2/rxjs.umd.min.js"></script>
<script>const {merge, of} = rxjs; const {startWith, endWith} = rxjs.operators;</script>

现在，如果上面的三个 observable 都没有发出值，我不想输出 >>> 和 <<<。所以startWith 和endWith 只有在merge(foo$, bar$, baz$) 实际发出一个值时才能“运行”。

为了模拟我使用过滤功能拒绝foo$、bar$ 和baz$ 发出的所有值。

const foo$ = of('foo').pipe(filter(() => false));
const bar$ = of('bar').pipe(filter(() => false));
const baz$ = of('baz').pipe(filter(() => false));

merge(foo$, bar$, baz$).pipe(startWith('>>>'), endWith('<<<')).subscribe(str => {
  console.log(str);
});

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.2/rxjs.umd.min.js"></script>
<script>const {merge, of} = rxjs; const {startWith, endWith, filter} = rxjs.operators</script>

但是，正如您在输出中看到的那样，startWith 和 endWith 都发出了它们的值，即使 merge() 没有产生任何值。

问题：如果 observable 没有发出单个值，如何防止 startWith 和 endWith 执行？

Answer 1

来自startWith的文档：

返回一个 Observable，它在开始发射源 Observable 发射的项目之前发射您指定为参数的项目。

所以startWith 和endWith 将始终运行。

我不知道您的预期结果应该是什么，但如果您只想为每个发出的值连接字符串，您可以使用 map 或 switchMap 运算符。

编辑： 使用map 连接每个值的示例：

const foo$ = of('foo').pipe(filter(() => false));
const bar$ = of('bar').pipe(filter(() => false));
const baz$ = of('baz').pipe(filter(() => false));

merge(foo$, bar$, baz$).pipe(map(v => `>>>${v}<<<`)).subscribe(str => {
  console.log(str);
});

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.2/rxjs.umd.min.js"></script>
<script>const {merge, of} = rxjs; const {startWith, endWith, filter, map} = rxjs.operators</script>

Answer 2

实现这一目标的一种方法是使用 materialize-dematerialize operators 对：

source$.pipe(
  // turn all events on stream into Notifications
  materialize(),

  // wrap elements only if they are present
  switchMap((event, index) => {
    // if its first event and its a value
    if (index === 0 && event.kind === 'N') {
      const startingNotification = new Notification('N', '>>>', undefined);
      return of(startingNotification, event);
    }

    // if its a completion event and it not a first event
    if (index > 0 && event.kind === 'C') {
      const endingNotification = new Notification('N', '<<<', undefined);
      return of(endingNotification, event);
    }

    return of(event);
  }),

  // turn Notifications back to events on stream
  dematerialize()
)

在操场上玩这个代码：
https://thinkrx.io/gist/5a7a7f1338737e452ff6a1937b5fe05a
为方便起见，我还在那里添加了一个空的错误源

希望对你有帮助

Answer 3

第一个条件很简单。你可以在第一个发射前加上concatMap：

mergeMap((v, index) => index === 0 ? of('>>>', v) : of(v))

第二个条件更棘手。你想要基本上与defaultIfEmpty相反。我想不出任何简单的解决方案，所以无论如何我可能会使用endWith，如果它是第一个也是唯一一个发射（这意味着源刚刚完成而没有发射任何东西），则忽略发射：

endWith('<<<'),
filter((v, index) => index !== 0 || v !== '<<<'),

完整示例：

const foo$ = of('foo');//.pipe(filter(() => false));
const bar$ = of('bar');//).pipe(filter(() => false));
const baz$ = of('baz');//.pipe(filter(() => false));

merge(foo$, bar$, baz$).pipe(
  mergeMap((v, index) => index === 0 ? of('>>>', v) : of(v)),
  endWith('<<<'),
  filter((v, index) => index !== 0 || v !== '<<<'),
).subscribe(console.log);

现场演示：https://stackblitz.com/edit/rxjs-n2alkc?file=index.ts

Answer 4

您可以将合并后的 observable 分配给一个变量，从流中获取第一个元素，然后在发出至少一个值时映射到要执行的 observable。

const foo$ = of('foo').pipe(filter(() => false))
const bar$ = of('bar').pipe(filter(() => false))
const baz$ = of('baz').pipe(filter(() => false))

const merged$ = merge(foo$, bar$, baz$);
merged$.pipe(
  take(1),
  switchMap(() => merged$.pipe(
    startWith('>>>'),
    endWith('<<<')
  ))
).subscribe(console.log);

https://stackblitz.com/edit/rxjs-phspsc

Answer 5

我看到所有答案都是组合运算符完成的，但是仅使用运算符完成这个解决方案有点棘手，为什么不直接创建自己的 observable？我认为这个解决方案是最容易理解的。 没有隐藏的聪明之处，没有复杂的运算符组合，没有订阅重复......

解决方案：

const foo$ = of('foo')//.pipe(filter(() => false));
const bar$ = of('bar')//.pipe(filter(() => false));
const baz$ = of('baz')//.pipe(filter(() => false));

function wrapIfOnEmit$(...obs) {
  return new Observable(observer => {
    let hasEmitted;
    const subscription = merge(...obs).subscribe((data) => {
      if (!hasEmitted) {
        observer.next('>>>');
        hasEmitted = true;
      }
      observer.next(data);
    },
    (error) => observer.error(error),
    () => {
      if (hasEmitted) observer.next('<<<');
      observer.complete();
    })
    return () => subscription.unsubscribe();
  })
}

wrapIfOnEmit$(foo$, bar$, baz$).subscribe(console.log);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/6.5.2/rxjs.umd.min.js"></script>
<script>const {merge, of, Observable} = rxjs; const {filter} = rxjs.operators</script>

希望这会有所帮助！