Emit last value on stream silence

[godind]

I’m drawing observable stream of values on a chart over time. The chart needs to refreshes on a fix schedule. Say 500ms. I use sampleTime(500) to have a consistent flow but sometimes there is no new data send in that 500ms period and so nothing is emitted and the chart has blanks.

How can I listen to Observable silences and re-emit the value that was received?

The logic would be: use sampleTime(500) values, if no new data in the last 500ms, send the last received value.

thanks

[demensky]

I can suggest this implementation:

source$
  .pipe(switchMap((value) => interval(1000).pipe(map(() => value))))
  .subscribe(console.log);

[demensky]

As a custom operator:

function sampleInterval<T>(period: number): MonoTypeOperatorFunction<T> {
  return switchMap((value) => interval(period).pipe(map(() => value)));
}

source$.pipe(sampleInterval(1000)).subscribe(console.log);

[godind]

Wow fantastic! How would I integrate this with sampleTime so it uses the last sample?
source$.pipe(sampleInterval(1000),sampleTime(1000)).subscribe( ... )

This way it sometimes skip a second. Appears to be the first sequence in the sampleInterval fill in values. Or is it emitting value regardless if sampleTime sends a new value?

• value: 5.97 - 43' 564"
• value: 6.33 - 44' 562"
• value: 6.69 - 45' 562"
• MISSING
• value: 7.2 - 47' 562"
• value: 7.77 - 48' 562"
• value: 7.2 - 49' 561"
• value: 7.2 - 50' 562"
• MISSING
• value: 7.2 - 52' 562"
• MISSING
• value: 7.2 - 54' 562"
• value: 6.69 - 55' 561"

[demensky]

You need to use it instead of sampleTime. Well, or I misunderstood the task.

[demensky]

Based on your example, it looks like this implementation:

function sampleInterval<T>(period: number): MonoTypeOperatorFunction<T> {
  return (source) =>
    interval(period).pipe(withLatestFrom(source, (_, value) => value));
}

[godind]

I'm probably not explaining well! Yes I want to combine it with sampleTime so when sampleTime does not emit a value, because there was nothing to sample in the time period, I can emit the previous sampleTime value.

I have this:
source$.pipe(sampleTime(1000)).subscribe( ...process the values )

This gives me the last value received in the last 1000ms. It's a consistent flow as needed for my chart. So I receive:

• value = 4 - 1 sec
• value = 6 - 2 sec
• value = 4.5 - 3 sec
• ...

But if no values are received in the last 1000ms, which does occur sometimes, I get "holes" in the data. Like so:

• value = 4 - 1 sec
• nothing
• value = 4.5 - 3 sec
• ...

So I want to fill this gap with the last emitted SampleTime value. Such as:

• value = 4 - 1 sec
• Filled with value from sec 1 sampleTime : value = 4 - 2 sec
• value = 4.5 - 3 sec
• ...

Makes sens?

[godind]

Oh. You mean no need for sampleTime any more...! I tried with source$.pipe(sampleInterval(1000)).subscribe( ) and it skips some interval. It seems that if I do get value, or many values, during the interval - not sure, it's skips.

• value = 5.3 - 40' 489"
• value = 4.84 - 41' 497"
• Missing a few seconds
• value = 5.35 - 44' 508"
• value = 5.4 - 45' 540"
• value = 4.94 - 46' 547"
• value = 5.3 - 47' 557"
• here also
• value = 5.61 - 49' 563"
• value = 5.56 - 50' 574"
• value = 5.45 - 51' 582"
• then here
• value = 5.14 - 54' 591"

[demensky]

What about this option?

[godind]

@demensky Bingo! Thank you so much.

Would be great is sampleTime had a config, similar to ThrottleTime, where we could set a send last value if no new value: true/false, as well as trailing/leading options. I guess many could have this...

Thanks again!