RxJS - subscribe only once but do not complete Observable

Question

imagine situation when you have some Observable that contains data that changes in real time, example below...

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

userData: Observable<User>

This userData observable is used in component to show some data that changes in real time. e.g.

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

Now I want to insert some data to database according to current data in userData observable. Here is the function

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

Question

Is this a correct solution to unsubscribe inside subscription to avoid multiple insertion into database? Is there a different/better way to do this?

This is just minimalistic example, if you have some questions or my explanation is poor, let me know in comments and I will update my question. Thank you

Answer 1

You can use the first operator:

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

This will automatically complete (and therefore unsubscribe) after the first value has been emitted.

Note that you should ensure that it emits at least once before completing as otherwise an error will be thrown. If you can't ensure this, you can use take(1) instead:

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

Note that this doesn't actually modify the userData observable directly, so other subscriptions to it will continue emitting regardless. This is because operators in rxjs do not modify observables but instead return a new observable.