用事件驅動超時清除Rx.Observable bufferCount？

Question

我正在使用rxjs 5.0：

如何在此緩衝區上設置超時值。所以當5秒內沒有發生keyup事件時，它會清除bufferCount（11）？

var keys = Rx.Observable.fromEvent(document, 'keyup'); 
var buffered = keys.bufferCount(11,1); 
buffered.subscribe(x => console.log(x)); 

Answer 1

您可以附加一個timeoutWith，這可能會在一定的時間（在你的情況5秒）後返回新鮮buffered。

const keys$ = Rx.Observable.fromEvent(document, "keyup") 
 
    .map(ev => ev.keyCode|| ev.which); // this is just to have a readable output here in the SO-console 
 
    
 
const buffered$ = keys$ 
 
    .bufferCount(3,1) // replaced your 11 with 3 for easy demonstration 
 
    .timeoutWith(2000, Rx.Observable.defer(() => { // replaced 5 with 2 seconds (easier to test here) 
 
    console.log("New Buffer!"); 
 
    return buffered$; 
 
    })); 
 

 
buffered$.subscribe(console.log);

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

---

作爲改進，這可能會更加強，只有開始對第一行程流，否則，我們將有一個恆定的超時運行（不是關鍵，但仍然可以防止）。

const keys$ = Rx.Observable.fromEvent(document, "keyup") 
 
    .map(ev => ev.keyCode|| ev.which); // this is just to have a readable output here in the SO-console 
 

 
const buffered$ = keys$ 
 
    .take(1) 
 
    .switchMap(firstKey => { 
 
    console.log("New Buffer!"); 
 
    return keys$ 
 
    .startWith(firstKey) 
 
    .bufferCount(3,1) // replaced your 11 with 3 for easy demonstration 
 
    .timeoutWith(2000, Rx.Observable.defer(() => buffered$)); // replaced 5 with 2 seconds (easier to test here) 
 
    }); 
 

 
buffered$.subscribe(console.log);

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

Answer 2

我有使用window和switchMap()另一個（可能更容易理解）解決方案：

var keys = Rx.Observable.fromEvent(document.getElementById('myinput'), 'keyup') 
    .map(event => event.keyCode) 
    .share(); 

var buffered = keys 
    .window(keys.debounceTime(5000)) 
    .switchMap(observable => observable.bufferCount(5, 1)) 
    .filter(buffer => buffer.length === 5); 

buffered.subscribe(x => console.log(x)); 

觀看演示：https://jsbin.com/cakoru/17/edit?js,console,output

當你不輸入至少5秒，window()運營商創建一個新的Obser在switchMap()內部訂閱並與新的.bufferCount()運營商鏈接。

Answer 3

以下是我會做：

const keys$ = Rx.Observable.fromEvent(document, 'keyup').map(ev => ev.keyCode|| ev.which); 

keys$ 
    .debounceTime(5000) 
    .startWith({}) 
    .switchMap(x => keys$.bufferCount(11, 1)) 
    .subscribe(x => console.log(x)); 

在這裏，我們已經得到了產生一個值，每次打字停止五秒鐘（與虛值，拉開序幕），其switchMaps成bufferCount流。