发射一次，然后每秒发射一次，直到值发生变化答案

【问题标题】：Emit once and then every second thereafter until value changes发射一次，然后每秒发射一次，直到值发生变化
【发布时间】：2018-12-31 18:36:33
【问题描述】：

我有这个工作。收到套接字消息后，立即发出一个值，此后每隔一秒发出一个值（增加年龄），直到再次收到套接字。

但是，无论是否收到套接字，我都希望它每秒发出一次。所以它会开始每秒发射一次，但是当收到套接字时，属性会更改为新的，并且它们会每秒发射一次。

不知道该怎么做。

updated: Observable<TargetDevice>;

this.updated = socketService.onMessage.pipe(
  filter(
    message =>
      message.messageType === SocketIoMessageType.Device &&
      message.data.id === this.id
  ),
  map((message: SocketIoMessage) => <Device>message.data),
  tap(d => this.setProps(d)),
  switchMap(d =>
    timer(0, 1000).pipe(
      tap(tick => (this.age = d.age + tick)),
      map(() => this)
    )
  )
);

【问题讨论】：

所以你想要一个不是来自套接字的起始值，只是开始你的一秒间隔还是我错过了什么？

标签： rxjs rxjs6

【解决方案1】：

您想使用combineLatest 和startWith 来实现您想要的行为：

combineLatest(
    socketService.onMessage.pipe(
        startWith(DEFAULT_MESSAGE)
    ),
    timer(0, 1000)
).pipe(
    //...
)

【讨论】：

【解决方案2】：

解决问题的一个可能方法是创建beforeSocketMessage$ 流，并使用takeUntil 运算符对其进行限制。

const beforeSocketMessage$ = interval(1000).pipe(mapTo('beforeSocketMessage'), takeUntil(socketService.onMessage));


const message$ = socketService.onMessage.pipe(
    filter(/* you filter*/),
    map((message: SocketIoMessage) => <Device>message.data),
    tap(d => this.setProps(d)),
    switchMap((data) => interval(0, 1000).pipe(mapTo(data))),
    tap(tick => (this.age = d.age + tick)),
    mapTo(this)
)

this.update = merge(beforeSocketMessage$, message$);

stackblitz.com 上的简化版本：https://stackblitz.com/edit/rxjs-jwg2cb?devtoolsheight=60

【讨论】：

【解决方案3】：

我想你想要这样的东西。你对switchMap是正确的。很酷的一点是，它会在发出下一个值后取消订阅源 observable。

socketService.onMessage.pipe(
    switchMap(value => interval(1000).map(() => value))
);

在 RxJs Playground here查看它

【讨论】：

实际上比这要复杂一些，见编辑。