Not your question, but I needed the same functionality

import { takeWhileInclusive } from 'rxjs-take-while-inclusive'
import { of, interval, race, throwError } from 'rxjs'
import { catchError, timeout, mergeMap, delay, switchMapTo } from 'rxjs/operators'

const defaultMaxWaitTimeMilliseconds = 5 * 1000

function isAsyncThingSatisfied(result) {
  return true
}

export function doAsyncThingSeveralTimesWithTimeout(
  doAsyncThingReturnsPromise,
  maxWaitTimeMilliseconds = defaultMaxWaitTimeMilliseconds,
  checkEveryMilliseconds = 500,
) {
  const subject$ = race(
    interval(checkEveryMilliseconds).pipe(
      mergeMap(() => doAsyncThingReturnsPromise()),
      takeWhileInclusive(result => isAsyncThingSatisfied(result)),
    ),
    of(null).pipe(
      delay(maxWaitTimeMilliseconds),
      switchMapTo(throwError('doAsyncThingSeveralTimesWithTimeout timeout'))
    )
  )

  return subject$.toPromise(Promise) // will return first result satistieble result of doAsyncThingReturnsPromise or throw error on timeout
}

Example

// mailhogWaitForNEmails
import { takeWhileInclusive } from 'rxjs-take-while-inclusive'
import { of, interval, race, throwError } from 'rxjs'
import { catchError, timeout, mergeMap, delay, switchMap } from 'rxjs/operators'

const defaultMaxWaitTimeMilliseconds = 5 * 1000

export function mailhogWaitForNEmails(
  mailhogClient,
  numberOfExpectedEmails,
  maxWaitTimeMilliseconds = defaultMaxWaitTimeMilliseconds,
  checkEveryMilliseconds = 500,
) {
  let tries = 0

  const mails$ = race(
    interval(checkEveryMilliseconds).pipe(
      mergeMap(() => mailhogClient.getAll()),
      takeWhileInclusive(mails => {
        tries += 1
        return mails.total < numberOfExpectedEmails
      }),
    ),
    of(null).pipe(
      delay(maxWaitTimeMilliseconds),
      switchMap(() => throwError(`mailhogWaitForNEmails timeout after ${tries} tries`))
    )
  )

  // toPromise returns promise which contains the last value from the Observable sequence.
  // If the Observable sequence is in error, then the Promise will be in the rejected stage.
  // If the sequence is empty, the Promise will not resolve.
  return mails$.toPromise(Promise)
}

// mailhogWaitForEmailAndClean
import { mailhogWaitForNEmails } from './mailhogWaitForNEmails'

export async function mailhogWaitForEmailAndClean(mailhogClient) {
  const mails = await mailhogWaitForNEmails(mailhogClient, 1)

  if (mails.count !== 1) {
    throw new Error(
      `Expected to receive 1 email, but received ${mails.count} emails`,
    )
  }

  await mailhogClient.deleteAll()

  return mails.items[0]
}

We also have the same use case and the below code works pretty good.

import { timer, Observable } from "rxjs";
import { scan, tap, switchMapTo, first } from "rxjs/operators";

function checkAttempts(maxAttempts: number) {
  return (attempts: number) => {
    if (attempts > maxAttempts) {
      throw new Error("Error: max attempts");
    }
  };
}

export function pollUntil<T>(
  pollInterval: number,
  maxAttempts: number,
  responsePredicate: (res: any) => boolean
) {
  return (source$: Observable<T>) =>
    timer(0, pollInterval).pipe(
      scan(attempts => ++attempts, 0),
      tap(checkAttempts(maxAttempts)),
      switchMapTo(source$),
      first(responsePredicate)
    );
}

if the number of attempts has reached the limit, an error is thrown which results in the output stream being unsubscribed. Moreover, you only make http requests until the given condition defined as responsePredicate is not met.

Exemplary usage:

import { of } from "rxjs";

import { pollUntil } from "./poll-until-rxjs";

const responseObj = { body: { inProgress: true } };
const response$ = of(responseObj);
// this is to simulate a http call
response$
  .pipe(pollUntil(1000, 3, ({ body }) => !body.inProgress))
  .subscribe(({ body }) => console.log("Response body: ", body));

setTimeout(() => (responseObj.body.inProgress = false), 1500);

Angular / typescript rewritten solution from above:

export interface PollOptions {
  interval: number;
  timeout: number;
}

const OPTIONS_DEFAULT: PollOptions = {
  interval: 5000,
  timeout: 60000
};

@Injectable()
class PollHelper {
  startPoll<T>(
    pollFn: () => Observable<T>, // intermediate polled responses
    stopPollPredicate: (value: T) => boolean, // condition to stop polling
    options: PollOptions = OPTIONS_DEFAULT): Observable<T> {
    return interval(options.interval)
      .pipe(
        exhaustMap(() => pollFn()),
        first(value => stopPollPredicate(value)),
        timeout(options.timeout)
      );
  }
}

Example:

pollHelper.startPoll<Response>(
  () => httpClient.get<Response>(...),
  response => response.isDone()
).subscribe(result => {
  console.log(result);
});