Создание строки данных R строка за строкой

Итак, ваша проблема в том, что stream продолжает излучать даже после того, как messages пуст.

Вы можете использовать takeWhile для завершения потока:

const stream = interval(1000).pipe(
  map((): Message => messages.pop()),
  takeWhile(Boolean),
);

Когда messages пусто, он возвращает undefined, то есть false, когда он превращен в логическое значение, поэтому takeWhile(Boolean) завершит поток.