Pull sampler when input source emits

[mperktold]

First of all, I wanted to say I like how this operator allows to interfrate pullable and listenable sources to form powerful complex pipelines. In this sense, I propose a feature request which would make it even more powerful.

With the current implementation, we can provide a listenable sampler that tells us when to pull, e.g. an interval that emits every second.

However, there is no easy way to pull the source based on its emissions, like pulling it whenever it another item whenever we get one etc.

But this can easily be achieved by pulling the sampler whenever the input emits, possible passing along the data as well.

This simple (andfully backwards-compatible) change would open some cool possibilities. For example, you could can pull another item after each emission, but first waiting a certain amount of time:

pipe(
  range(1, 100),
  pullWhen(pipe(
    fromIter(function* () {       // some infinite pullable
      while (true) yield 0;
    }),
    switchMap(_ => timer(1000)),
    startWith(0)                  // start immediately
  )),
  observe(x => console.log(x))    // logs 1, 2 ... with pauses of 1 sec in between
)

Another possibility would be to wait for n emissions before pulling n more items.

If you want, I can create a pull request for this.

[Andarist]

Hm, I don't think your sample code would work, but I could have missed something. The inner pipe isn't being pulled, so actually fromIter would never emit.

Could you maybe also describe your use case a little bit more? I'm having slightly hard time to understand what you are trying to achieve. Do you want to pull from source A based on emissions from source A?

[mperktold]

Hm, I don't think your sample code would work, but I could have missed something. The inner pipe isn't being pulled, so actually fromIter would never emit.

Exactly; the inner source is never pulled, so it doesn't work. This issue is about changing pullWhen to make this work.

Do you want to pull from source A based on emissions from source A?

Not quite. Consider a pipeline of the following form.

const output = pullWhen(puller)(input);

So input is the upstream pullable callbag, sampler is the inner callbag which decides when to pull input, and output is the result of applying pullWhen(sampler) to input.

The idea is to inform sampler about emissions from input. This would allow sampler to cause a pull after each emission, similar to callbag-pump, but more flexible, including the following features:

• Delaying pulls by a certain amount of time (like in the example above).
• Pulling another item only after certain items, using operators like filter or distinctUntilChanged in the sampler.

I'll create a pull request with this feature. That will hopefully clearify what I mean.

[Andarist]

I'm really curious about how you want this to be implemented - gonna wait for a PR demonstrating it. I'm still not sure if this should be a part of this operator though, but I'm open for further discussion.

Seems to me that it could be implemented on top of existing operators without extending this one.