Hi i'm having a hard time converting this RxJS cod...

# flow

[althaf]

Hi i'm having a hard time converting this RxJS code to Kotlin flow

Some how in java script after the buffer we are getting the stream events as array , however in flow i'm not receiving it has an array of events. Is this because buffer() behaves differently from bufferWhen in RxJS ?

var button = document.querySelector('.button');
var label = document.querySelector('h4');

var clickStream = Rx.Observable.fromEvent(button, 'click');

var doubleClickStream = clickStream
  .bufferWhen(() => clickStream.debounceTime(250))
  .map(arr => arr.length)
  .filter(len => len === 2);

doubleClickStream.subscribe(event => {
  label.textContent = 'double click';
});

doubleClickStream
  .delay(1000)
  .subscribe(suggestion => {
    label.textContent = '-';
  });

[althaf]

Working example is available here
https://embed.plnkr.co/github/eggheadio-projects/introduction-to-reactive-programmin[…]-stream-of-double-clicks-in-rxjs?show=lib%2Fscript.js,preview

[Nick Allen]

Those methods are not remotely similar. Flow.buffer affects back-pressure. RxJs's bufferWhen, groups items together.

I think you'll need to build your own operator, something like:

private object MySignal
fun <T, S> Flow<T>.myBufferWhen(signal: Flow<S>): Flow<List<T>> = callbackFlow<List<T>>{
    val merged = merge(this as Flow<Any?>, signal.map { MySignal})
    var nextList = mutableListOf<T>()
    merged.collect {
        if (it is MySignal) {
            if (nextList.isNotEmpty()) {
                send(nextList)
                nextList = mutableListOf<T>()
            }
        } else {
            nextList.add(it as T)
        }
    }
    if (nextList.isNotEmpty()) {
        send(nextList)
    }
}

(not tested, just typed this out in slack)