.bufferWithTimeOrCount (& .bufferWithCount?)

[rpominov]

in RxJS in Bacon

We probably need this.

We also lack .bufferWithTime and .bufferWithCount, but those are relatively easy to replicate using existing API:

// .bufferWithTime
foo.bufferBy(Kefir.interval(1000))

// .bufferWithCount
let i = 0
foo.bufferWhile(() => i++ % n !== 0)

Although having to store i state somewhere for .bufferWithCount also not very cool. Maybe we need this one too.

PRs are welcome! :dancer:

[vikikamath]

:+1:

[mcmathja]

I'd be interested in taking this. To clarify, expected behavior is that timer is reset to 0 after a flush initiated by count, but otherwise should run continuously between activation and deactivation?

[rpominov]

@mcmathja Yeah, I think timer should be reset to 0 on each flush (no matter what initiated it), and we should start timer on activation, and stop & reset it on deactivation.

[mcmathja]

I've got something up and running, but I had a question about how Kefir handles inheritance: this definitely seems like a single-activation pattern, but we need to override _onActivation and _onDeactivation to properly handle the interval. There's no straightforward way to call up to the pattern's implementation of these methods once they're overridden - I considered using time-based instead, but it can only return streams, and adding a super implementation to inherits seems to call for a big piece of refactoring. In to-property and most of the two-sources streams, the activation / deactivation methods defined in the patterns are simply copied over and reimplemented with local modifications - would that be the preferred approach to doing this?

[rpominov]

We should probably do it same way as in to-property here. I know it's not the perfect solution, but easiest one and shouldn't make much harm. I wouldn't use inheritance/mixins/etc at all if it wouldn't be good for performance :sweat_smile:

[mcmathja]

Haha, all right! :stuck_out_tongue_winking_eye:

One other question, actually. I was reading through Bacon's source for this and they handle scheduling somewhat differently. Assuming the buffer is never flushed because the count limit is reached, the processing order ends up being:

1. Don't set any timeout until the first event is received.
2. When a timeout is fired,
   1. if there are values buffered, flush and reset the timer.
   2. If there are no buffered values, unset the timer and return to initial state.

The logic behind 2.i seems weird to me, but 1 made me rethink this a little. The issue with the current approach is that the interval specified runs continuously from activation, which means you'll get inconsistent behavior depending on when during the current cycle you happen to process your first value - it could be 200 ms to the next flush or it could be 1000 ms. That seems like it would throw off users, and the timing doesn't really have any semantic value in that initial case or in cases where the interval has already elapsed since the previous emission. A more predictable approach would be:

1. Start with no timeout.
2. When an event is received, if the buffer is empty, start timeout.
3. On flush (for any reason), clear timeout.

This is essentially saying give me bufferWithCount, but don't let any events in the buffer get older than the specified wait, which seems to be the primary use case for this sort of operation. Would that be more useful?

[rpominov]

Hm, not sure what is more preferable semantics here. I can imagine people wanting something like "give me stream with buffer that flushed at least every N milliseconds or when there is M items in it", in that case we want to start timer on activation.

But I understand the issue with inconsistent behavior. Basically time of first subscription will have side effect here — will potentially affect time of first flush. But unfortunately we already have this issue in some places, so it won't be new.

I don't have strong feeling either way, but "start timer on activation" (i.e. "flushed at least every N milliseconds") behavior seems to me like easier to grasp for users.

[rpominov]

I've just released 3.2.0 with this methods. Thanks again @mcmathja and @elsehow!