How to throttle a stream to get values as early as possible, while getting at most one value every 500ms

[Recursing]

Let's say I have a stream of events; I want to get the first one immediately, and after get at most one every 500ms.
So, if more than 500ms have passed since the previous one: send the event immediately, otherwise wait whatever ms are left and then send the latest one.

An ugly implementation that seems to work is:

let lastTime = 0;
myObserver$
  .pipe(
    debounce(() =>
      timer(Math.max(0, lastTime + 500 - new Date().getTime())),
    ),
    tap(() => {
      lastTime = new Date().getTime();
    }),
    ...

But of course it's very ugly.
Using

throttleTime(500, asyncScheduler, {leading: true, trailing: true})

almost works. But if I get events at times 0ms, 499ms, 501ms, 502ms, 503ms, I will get both 499ms and 501ms immediately, which is undesirable.

Is there a prettier way than manually keeping track of lastTime and using a debounce with a variable timer?
I think this would be a pretty common use case, it could be seen as some kind of "eager debounce": you (usually) keep responsiveness while throttling expensive operations

Marble diagram, vertical ticks are 500ms apart:

[cartant]

When I was going through the issues, a few weeks ago, inspectTime - which was removed before I started using RxJS - popped up a couple of times. If that operator were available, would it do what you want? See #4429

[Recursing]

I'm not sure I fully understand all the nuances, but it really seems exactly what I am looking for!

Are there obvious problems/differences in my current approach (the ugly implementation above)? Do you think it might be worth it to bring inspectTime back, given there seems to be some need for it?

[cartant]

Given that it's the third time I've seen this recently, I think it's worth discussing. At least to see whether it can be composed from what's already in the core. I'll have a think about it.

[josepot]

If I understood the problem correctly, I think that another way to create an operator that behaves like that would be something like:

export const customThrottle = <T>(time: number): MonoTypeOperatorFunction<T> =>
  connect((hotSource$) =>                                                
    concat(                                                              
      hotSource$.pipe(take(1)),                                          
      hotSource$.pipe(                                                   
        auditTime(time),                                                 
        takeUntil(hotSource$.pipe(switchMapTo(timer(time)))),            
      ),                                                                 
    ).pipe(repeat()),                                                    
  )                                                                      

If you are using RxJS 6, then replace connect with publish. Also, sorry but I'm really bad at coming up with good names 😅

[josepot]

You are absolutely right! Sorry about that, I misunderstood how auditTime works.

There are probably easier ways to do this, but check this out:

 const NIL = {}
                            
 export const customThrottle = <T>(time: number): MonoTypeOperatorFunction<T> =>
   publish((hotSource$) => {                                                    
     const waitAndEmit$ = timer(time).pipe(                                     
       withLatestFrom(hotSource$.pipe(startWith(NIL))),                         
       pluck(1),                                                                
     )                                                                          
                                                                                
     return concat(                                                             
       hotSource$.pipe(take(1)),                                                
       waitAndEmit$.pipe(                                                       
         expand((value) => (value === NIL ? [] : waitAndEmit$)),                
         filter((value): value is T => value !== NIL),                       
       ),                                                                       
     ).pipe(repeat())                                                           
   })                                                                           

I've tried this marble test and it passes:

// TIME MARKS: '   x    x    x    x    x  '                          
const source = "  -a-b---c-d--e-f--------g"                          
const expected = "-a----b----d----f------g"                          
expectObservable(cold(source).pipe(customThrottle(5))).toBe(expected)

[Recursing]

Very impressive! 🤯
In the next few days I'll try to decode what it does, indeed it seems to pass some basic testing.
Will certainly learn a bit about rxjs trying to understand what is going on there

[josepot]

Will certainly learn a bit about rxjs trying to understand what is going on there

This is how I thought about the problem:

1. Let's start with an observable that the only thing that it does is to emit the first value of the source and then it completes.
2. Next let's have an observable that starts a timer with the latest emitted value of the source, also if the source did not emit let's signal that with some token (NIL). This observable always emits once and then it completes.
3. a: If the previous observable did emit an actual value (not NIL), then lets go back to step 2
   b: If it emitted NIL, then let's filter out that internal token and let's go back to step 1

For this to work properly, we need to make sure that we are working with a multicasted (hot) verion of the source observable, that's why we are using publish (or connect if you are on v7).

Actually, after writing down the explanation I came up with a slightly simpler implementation:

const NIL = {}                                                                 
export const customThrottle = <T>(time: number): MonoTypeOperatorFunction<T> =>
  publish((hotSource$) => {                                                    
    const firstValue$ = hotSource$.pipe(take(1))                               
                                                                               
    const waitAndEmitLatestValue$ = timer(time).pipe(                          
      withLatestFrom(hotSource$.pipe(startWith(NIL))),                         
      pluck(1),                                                                
    )                                                                          
                                                                               
    const waitAndEmitLatestValueRecursivelyUntilNil$ =                         
      waitAndEmitLatestValue$.pipe(                                            
        repeat(),                                                              
        takeWhile((value): value is T => value !== NIL),                       
      )                                                                        
                                                                               
    return concat(                                                             
      firstValue$,                                                             
      waitAndEmitLatestValueRecursivelyUntilNil$                               
    ).pipe(repeat())                                                           
  })                                                                           

[confraria]

what do you think about this one?
I believe it works the same way.

let ref = Date.now();

const specialThrottle = timeInMs =>
  connect(source =>
    concat(
      source.pipe(first()),
      source.pipe(
        switchMap(v => concat(of(v), interval(timeInMs).pipe(mapTo(v)))),
        throttleTime(timeInMs, asyncScheduler, {
          trailing: true,
          leading: false
        })
      )
    )
  );

const source = concat(of(-1), interval(100).pipe(take(100)), interval(2000));

source.pipe(specialThrottle(500)).subscribe(a => {
  console.log(a, 'diff:', Math.round(Date.now() - ref));
  ref = Date.now();
});

https://rxjs-gh2s96.stackblitz.io/

from the original question I'm not sure if the repeated values when source is not emitting are required. if not they switchmap could be removed.

[backbone87]

https://stackblitz.com/edit/66olp5?file=index.ts

import { BehaviorSubject, MonoTypeOperatorFunction, timer } from 'rxjs';
import { audit, filter, tap } from 'rxjs/operators';

function inspectTime<T>(duration: number): MonoTypeOperatorFunction<T> {
  return (source$) => {
    const subject$ = new BehaviorSubject(true);

    return source$.pipe(
      audit(() => subject$.pipe(filter(Boolean))),
      tap(() => {
        subject$.next(false);
        timer(duration).subscribe(() => subject$.next(true));
      }),
    );
  };
}

[samal-rasmussen]

The behavior of throttleTime changed in rxjs 7 so it should now work as you expect, if I understand correctly. It will now ensure the spacing between throttles is always at least the throttled amount.

I've described the change with a passing test on Stackoverflow:
https://stackoverflow.com/questions/57059666/how-does-throttletime-operators-config-parameter-work-throttleconfig/67701965#67701965