Angular / RxJS：如何生成延迟值序列？

Question

我有一系列消息。 每隔N 秒，我需要从 Observable 发出一条消息。

我尝试了下一个代码。第一条消息是延迟发出的，我在订阅者中得到了它，但是按顺序生成其他消息被停止了。 如果我删除 .delay() 方法调用，我会得到所需的序列，订阅者会对每条消息做出反应，但消息之间没有时间段。我该如何解决？

import { Observable } from 'rxjs/Observable';
import { fromArray } from 'rxjs/observable/fromArray';
import 'rxjs/add/operator/delay';


@Injectable({
  providedIn: 'root'
})
export class MessageService {

  constructor() { }

  getIncomingMessagesStream(): Observable<string> {
    const messageTimeout = 2000;

    const messages = ['Hi there!', 'How are you?', 'That is awesome :)'];

    return fromArray(messages)
      .delay(messageTimeout);
  }
}

订阅者示例：

messageService.getIncomingMessagesStream()
    .subscribe(message => console.log('New incoming message ', message) );

Answer 1

你需要让链只有在前一个项目被延迟后才发出下一个项目：

import { from, of } from 'rxjs';
import { delay, concatMap } from 'rxjs/operators';

...

return from(messages).pipe(
  concatMap(item => of(item).pipe(delay(messageTimeout))),
);

顺便说一句，您正在结合 RxJS = 5.5 样式（操作符的可管道和原型样式）。最好只使用 RxJS 5.5 并避免使用import 'rxjs/add/operator/concatMap';。见https://github.com/ReactiveX/rxjs/blob/master/doc/pipeable-operators.md

2019 年 7 月：针对 RxJS 6 更新。

观看现场演示：https://stackblitz.com/edit/rxjs5-hibr4m?file=index.ts