@lcsga/operators

Breaking changes:

v3.0.0

• bufferWhile now puts the value at the moment when the predicate returns false, as the first element of the incoming new buffer, since while is an exclusive keyword. Also doing so mimics the takeWhile operator behavior that is also exclusive.
• In the predicate of bufferWhile, index now represents the i-th source emission that has happened since the subscription`.

v2.0.0

• bufferWhile now integrates the value at the moment when the predicate returns false, as the last element of the ongoing buffer, to mimic all of the already existing bufferXxx operators.

This package provides a set of custom RxJS operators, to extend the list of already built-in ones.

• debounceTimeMap: This operator extends the familly of of switchMap, mergeMap, concatMap and exhaustMap.

  debounceTimeMap<T, R>(
    project: (value: T, index: number) => ObservableInput<R>,
    dueTime: number
  ): OperatorFunction<T, ObservedValueOf<ObservableInput<R>>>

  | argument | type | description | | --------- | ------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------ | | project | (value: T, index: number) => ObservableInput<R> | A function that, when applied to an item emitted by the source Observable, returns an Observable. | | dueTime | number | The timeout duration in milliseconds, for the window of time required to wait for emission silence before emitting the most recent inner source value. |

  Here is a use case you could easily encounter yourself:

  You made a searchbar and you'd like to send a fetch request, while typing within it.

  To avoid sending many requests at the same time (possibly causing an issue, receiving the first response after the second one for example) you would need to cancel the previous one by using a switchMap.

  Another great thing you could and should do is, before sending any request, wait for a certain amount of time with no more input. To do so, you could use a debounceTime.

  Example:

  <input type="text" placeholder="Github username" />

  const input = document.querySelector<HTMLInputElement>('input');
  
  fromEvent<GithubUser>(input, 'keydown')
    .pipe(
      debounceTime(300),
      switchMap(() => fromFetch('https://api.github.com/users/' + input.value, { selector: (res) => res.json() }))
    )
    .subscribe(console.log);

  With those two operators, everything works as expected... or not!

  What happens if your timing is bad and you press another set of key, with the first one pressed after 301ms, then the others under 300ms each?

  => You will never go through the switchMap for a second time, as you could expect, which means that the previous request won't be canceled!

  Here comes the debounceTimeMap custom operator to the rescue!

  As you could probably guess, it simply is a combination of a debounceTime and a switchMap.

  Example:

  <input type="text" placeholder="Github username" />

  const input = document.querySelector<HTMLInputElement>('input');
  
  fromEvent<GithubUser>(input, 'keydown')
    .pipe(
      debounceTimeMap(
        () => fromFetch('https://api.github.com/users/' + input.value, { selector: (res) => res.json() }),
        300
      )
    )
    .subscribe(console.log);

  In the example above, if you are unlucky and press another key a little after 300ms, after a first request has been sent already, the previous request will still be canceled and you won't face any unexpected behavior!

• debounceMap: It works exactly like debounceTimeMap but with a durationSelector instead of a dueTime!

  debounceMap<T, R>(
    project: (value: T, index: number) => ObservableInput<R>,
    durationSelector: (value: T) => ObservableInput<any>
  ): OperatorFunction<T, ObservedValueOf<ObservableInput<R>>>

  | argument | type | description | | ------------------ | ------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------- | | project | (value: T, index: number) => ObservableInput<R> | A function that, when applied to an item emitted by the source Observable, returns an Observable. | | durationSelector | (value: T) => ObservableInput<any> | A function that receives a value from the source Observable, for computing the timeout duration for each source value, returned as an Observable or a Promise. |

  Example:

  fromEvent<GithubUser>(input, 'keydown')
    .pipe(
      debounceMap(
        () => fromFetch('https://api.github.com/users/' + input.value, { selector: (res) => res.json() }),
        () => timer(300)
      )
    )
    .subscribe(console.log);

  • bufferWhile: Buffers the source Observable values until the predicates turns false.

  bufferWhile<T>(predicate: (value: T, index: number) => boolean, inclusive: boolean = false): OperatorFunction<T, T[]>

  | argument | type | description | | ----------- | -------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | predicate | (value: T, index: number) => boolean | A function that evaluates each value emitted by the source Observable.Until the predicate returns false the buffer is updated with each incomming values. When it returns false the buffer is emitted, with the last value, to the output Observable, before being reset for the next ongoing values.The index parameter is the number i for the i-th source emission that has happened since the subscription, starting from the number 0. | | inclusive | boolean | Optional. Default is false.When set to true the value that caused predicate to return false will also be buffered. |

Example:

const source$ = of(1, 2, 3, 4, 5, 6);
const predicate = (nb) => nb !== 4;

// exclusive (default)
source$.pipe(bufferWhile(predicate)).subscribe(console.log);
// Outputs:
// [1, 2, 3]
// [4, 5, 6]

// inclusive
source$.pipe(bufferWhile(predicate, true)).subscribe(console.log);
// Outputs:
// [1, 2, 3, 4]
// [5, 6]