Emit once and then every second thereafter until value changes

Question

I have this working. On receipt of a socket message, a value is emitted immediately and every second thereafter (incrementing the age), until socket received again.

However, I want it to emit every second regardless of whether socket has been received. So it would start off emitting every seconds, but when socket received the properties would change to the new ones and they would be emitted every second.

Can't quite figure out what to do.

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) ) ) );

So you want a starting value which does not originate from the socket and just starts your one second interval or am I missing something? — M4R1KU
– M4R1KU, Commented Dec 31, 2018 at 7:21

ggradnig · Accepted Answer · 2018-12-31 08:50:17Z

You'd want to use combineLatest and startWith to achieve your desired behaviour:

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

Maksim Romanenko · Accepted Answer · 2018-12-31 10:36:33Z

One of possible solutions for you problem is to create beforeSocketMessage$ stream, and limit it with takeUntil operator.

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$);

Simplified version at stackblitz.com: https://stackblitz.com/edit/rxjs-jwg2cb?devtoolsheight=60

M4R1KU · Accepted Answer · 2018-12-31 06:40:02Z

I think you want something like that. You were right with switchMap. The cool thing about it is that it does unsubscribe from the source observable after the next value has been emitted.

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

Check it out on RxJs Playground here

Collectives™ on Stack Overflow

Emit once and then every second thereafter until value changes

3 Answers 3

Comments

Comments

1 Comment

Hot Network Questions

Collectives™ on Stack Overflow

3 Answers 3

Comments

Comments

1 Comment

Related