where would you put takeUntil RXJS in this code that polls a server?

Question

So the code below has been developed off the answer I got in this stack overflow question.

the code is designed to poll my server until either a condition on the server is true, or polling has occurred for a minute.

I understand I can stop the polling after a minute using the takeUntil RXJS function. However, I have no idea where in the code I would put it in. As every place I put it in that I thought it would go, the code has errored out.

I am also using this tutorial from the learnrxjs website https://www.learnrxjs.io/learn-rxjs/operators/filtering/takeuntil

You'll notice the first line of the startastream() function is

const endtimer = timer(60000);

This is the condition that would fill the takeUntil() argument. So takeUntil(endtimer)

    start$ = this.http.get(environment.shochat_content_creator_set_valid_stream_start).pipe(
    tap(() => console.log('Stream start'))
  );
  poll$ = this.http.get(environment.check_if_stream_is_active_on_mux).pipe(
    tap(() => (this.streamready = true)),
    catchError(error => {
      console.log(error);
      return EMPTY;
    })
  );

  startastream(){
    const endtimer = timer(60000);
    this.streampollsubscription = this.start$.pipe(
      switchMap(() => timer(0, 5000).pipe(
        tap(() => console.log('Polling every 5s')),
        mergeMap(() => this.poll$)
      ))
    ).subscribe();

  }

Answer 1

You can simply place it in the pipe after switchMap:

    this.streampollsubscription = this.start$.pipe(
      switchMap(() => timer(0, 5000).pipe(
        tap(() => console.log('Polling every 5s')),
        mergeMap(() => this.poll$)
      )),
      takeUntil(endtimer)
    ).subscribe();

Check out this StackBlitz.