Ako obmedziť udalosť mousemove v RxJs pri použití Angular 2 + TypeScript?

Som v procese učenia RxJs. Používam Angular2 rc3. Nasledujúci prúd funguje, ale dáva mi príliš veľa udalostí mousemove. Chcem ich spomaliť pomocou časovania (throttling) alebo iného riadiaceho prúdu. Ako to môžem urobiť?

Pohyb myši Stream bez obmedzenia

const mouseMove$ = Observable.fromEvent<MouseEvent>(document, 'mousemove');

mouseMove$.subscribe( x => console.log(x)); // works great, many {mouse position object} 's

Jednoduché riešenie: Použitie obmedzovania by malo vyzerať nejako takto:

const mouseMove$ = Observable.fromEvent<MouseEvent>(document, 'mousemove');
const latestMouseEventEvery1Second$ = mouseMove$.sample(1000);

latestMouseEventEvery1Second$.subscribe( x => console.log(x)); // error

- tento sample() operátor som si vzal odtiaľto: http://reactivex.io/documentation/operators/sample.html

Ale toto nefunguje v projekte Angular 2 - CLI. Dáva mi túto chybu:

***Argument of type 'number' is not assignable to parameter of type 'Observable<\any>'*** - notice i've put <\MouseEvent> when casting.

Ďalší silnejší spôsob, ako dosiahnuť rovnaký výsledok, si myslím, môže byť nasledujúci: Bolo by skvelé, keby sme mohli odosielať najnovší mousemove item na základe položiek prijatých z iného streamu. Akýkoľvek stream - vytvorený nami..

Napríklad:

Keď dostaneme nový prvok (1, 2, 3..) z prúdu eachSecond$ (=náš „riadiaci prúd“), - vysielame ďalej (do mouseMoveEachSecond$) - posledný prijatý prvok z prúdu mouseMove$.

const eachSecond$ = Observable.timer(0, 1000); // starts at 0 and gives 1,2,3 as values each 1000 milliseconds.
const mouseMove$ = Observable.fromEvent<MouseEvent>(document, 'mousemove');

const mouseMoveEachSecond$ = Observable.merge(eachSecond$,mouseMove$)
  .some magic operator I can't find()
  // this could work like this:
    // many mouse items are coming at a high rate from the mouse stream
    // only a few items (1,2,3..) are coming form the eachSecond$ stream
    //the logic: We send downstream **only one mouse item** (the last) for each new item that is coming form the eachSecond$ stream.
);

mouseMoveEachSecond$.subscribe( x => console.log(x)); 
// a { mouse position object } = mouse item - only when you move the mouse 
// and no more than one {mouse position object} per second}

Zvuk je jednoduchý. Zistil som, že implementovať toto bolo obzvlášť ťažké - podľa toho, čo viem, je to preto, že stav prúdu by mal byť nejako zdieľaný. Je potrebný spôsob, ako povedať, čo už bolo odoslané..

V mouseMoveEachSecond$prichádzajú položky jedna po druhej, postupne. Alebo číslo, alebo objekt. Musíte vedieť:

  • Kedy bol posledný časový predmet, keď ste dostali nový predmet myši.
  • A ktorý bol posledný mouse item, keď ste dostali nový time item.

Toto vám umožní:

  • neposielať myší predmet ďalej, keď sa myš na obrazovke nepohybuje.
  • neposielať mouse item ďalej, kým nebol z každého prúdu eachSecond$uvoľnený nový time item.

Toto je možné urobiť pomocou globálnych premenných. Ale nie je to spôsob, akým sa veci robia v RxJS.. Neviem o spôsobe, ako uložiť poslednú odoslanú položku a použiť ju v ďalšom kroku. Operátor reduce môže uchovať stav: od jednej uvoľnenej položky - po druhú, ale ako môžeme použiť túto vlastnosť na dosiahnutie tohto správania? Tieto prúdy musia byť považované za nekonečné. Hmm... mal by byť jednoduchší spôsob, ako tieto 2 prúdy skombinovať.. Možno je ešte príliš skoro.. Mám pocit, že môj mozog nechce myslieť v prúdoch :)

Takže tie 2 otázky sú jasne položené:

  1. aké sú niektoré spôsoby, ako obmedziť stream mousemove v Angular 2?
  2. Môžem použiť iný tok na kontrolu toho, kedy uvoľním položky mousemove? Ak áno, môže byť tento „riadiaci tok“ ľubovoľný tok, alebo sú nejaké obmedzenia?

Ďakujem veľmi pekne:)

Odpoveď č. 1

Si tak blízko! Ironicky je tvoj magický operátor je sample.

Stará verzia funkcie sample (z RxJS4) bola preťažená a mohla ako argument prijať číslo. V RxJS5 (tej, ktorá je dodávaná s Angular2) bola rozdelená, takže sú vlastne dva operátory sample a sampleTime. Ten druhý prijíma časový argument a vezme poslednú udalosť v danom časovom okne, keď emituje.

const mouseMoveEachSecond$ = mouseMove$.sampleTime(1000);

Prvý prijíma ako svoj jediný argument Observable a vysiela poslednú prijatú položku vždy, keď ovládací Observable niečo vysiela.

t.j.

const eachSecond$ = Observable.timer(0, 1000); // starts at 0 and gives 1,2,3 as values each 1000 milliseconds.
const mouseMove$ = Observable.fromEvent<MouseEvent>(document, 'mousemove');

const mouseMoveEachSecond$ = mouseMove$.sample(eachSecond$);

mouseMoveEachSecond$.subscribe( x => console.log(x));

Odpoveď č. 2

čo tak importovať throttleTime:

import { throttleTime } from 'rxjs/operators';

a potom ho použiť takto:

mousemouve$.pipe(
      throttleTime(5000)
    )