angular - RxJS - 只订阅一次但不完成 Observable

Question

想象一下当你有一些包含实时变化的数据的 Observable 的情况，下面的例子......

interface User {
   name: string;
   projectId: string;
   dataThatChangesALotInRealTime: Object;
}

userData: Observable<User>

这个 userData observable 在组件中使用来显示一些实时变化的数据。例如

<p>
{{ (userData | async)?.dataThatChangesALotInRealTime }}
</p>

现在我想根据 userData observable 中的当前数据向数据库中插入一些数据。这是函数

addToDatabase() {
  let sub = this.userData.subscribe(data => {
     this.exampleDatabase.doc(`test/${data.dataThatChangesALotInRealTime.id}`)
          .add({ test: 'hello'})
     sub.unsubscribe() // <- This
  })
}

问题

这是取消订阅内部订阅以避免多次插入数据库的正确解决方案吗？有没有不同/更好的方法来做到这一点？

这只是一个简单的示例，如果您有任何问题或我的解释不充分，请在评论中告诉我，我会更新我的问题。谢谢

Answer 1

您可以使用first 运算符:

this.userData.pipe(first()).subscribe(...);

这将在发出第一个值后自动完成(并因此取消订阅)。

请注意，您应该确保它在完成之前至少发出一次，否则将抛出错误。如果不能确保这一点，可以使用 take(1) 代替:

this.userData.pipe(take(1)).subscribe(...);

请注意，这实际上并没有直接修改 userData observable，因此不管怎样，对它的其他订阅将继续发出。这是因为 rxjs 中的运算符不会修改 observable，而是返回一个新的 observable。