Latest operator about wonka

[hanrelan]

Actually I figured out how to do it with scan by using None as the initial value and then extracting the Some(x) in the subscribe.
However it turns out this doesn't quite do what I want as it just delays the latest value by 1000ms. Is there a way to do something like debounce except instead of waiting for a period of N ms of silence, it is just silent for N ms and then emits the latest value? Essentially something like auditTime

[kitten]

I assume you’re in a web environment since you’re using delay?

You’re probably looking for the debounce and/or throttle operators :)

• https://wonka.kitten.sh/api/operators#debounce
• https://wonka.kitten.sh/api/operators#throttle

[hanrelan]

Thanks for the quick response!

Yes it's a web environment.

It's similar to throttle but not quite the same. With throttle the emit happens before the throttle period begins. I want it to happen after the throttle period. Similarly debounce is close but waits for silence which isn't what I want.

To illustrate the difference, imagine scraping the html whenever the DOM updates. Let's say you get 10 DOM updates within 1 second.
If you scraped at every update, you would do 10 scrapes. You'd get all the latest data but you'd be using a lot of CPU.
If you use throttle(1000), you would scrape at t = 0 second and that's it. So you would miss the latest data that existed at t = 1 second.
With auditTime(1000), you would only scrape at t = 1 second. This way you're minimizing the amount of CPU while still grabbing the latest data

[kitten]

Oh, okay, sorry for the goose chase for the right operator! 😅 Maybe the correct primitive is then sample? https://wonka.kitten.sh/api/operators#sample

You can pass it a notifier source. Every time that notifier emits it takes the latest value from the source that is passed into sample, so in your case you could do sample(interval(1000)).

It seems like that's exactly what Rx is doing internally as well? https://github.com/ReactiveX/rxjs/blob/master/src/internal/operators/auditTime.ts#L56

[hanrelan]

Oh yes, I think sample is what I want. Initially I thought that it would pull the latest item from the source regardless of if there was anything new, but I just tested it and it seems to stop if there's nothing new to pull.
For anyone else comes across this, my final implementation is:

source
|> Wonka.scan((. _acc, x) => Some(x), None)
|> Wonka.sample(Wonka.interval(1000))
|> Wonka.subscribe((. x) => {
     switch (x) {
     | None => ()
     | Some(x) => print_endline(string_of_int(x))
     }
   });

If you can think of a way to get rid of the Option type and the switch statement there I'd love to know how to do that. In the general case we don't necessarily have an identity type (eg. 0 for int) that we could pass to scan.

[kitten]

Oh, I see, so you're trying to emit the initial first value of the source and then sample values afterwards? You can probably write your own latest operator by providing a more complex notifier source to sample.

Essentially we'd want the notifier to emit when the first value of the source is emitted, so we could write that like so in TypeScript:

import {
  Operator,
  pipe,
  share,
  concat,
  take,
  interval,
  sample,
} from 'wonka';

const latest = <A>(time: number): Operator<A, A> => source => {
  const shared = share(source);
  const notifier = concat([
    pipe(shared, take(1)),
    interval(time)
  ]);

  return pipe(shared, sample(notifier));
};

So we're creating a latest operator, share the source and create a notifier that emits when the first item is emitted on the source and afterwards start an interval.

In Reason you'd write it like so (haven't tested this so there may be some errors in the following sample)

open Wonka;

let latest = (time: int) => (source: sourceT('a)): sourceT('a) => {
  let shared = share(source);
  let notifier = concat([|
    shared |> take(1),
    interval(time)
  |]);

  shared |> sample(notifier)
};

[hanrelan]

Cool thanks!