Kako ograničiti učestalost događaja mousemove u RxJs kada koristite Angular 2 + Typescript?

U procesu sam učenja RxJs-a. Koristim Angular2 rc3. Sljedeći stream radi, ali mi daje previše mousemove događaja. Želim ih usporiti koristeći time (throttling) ili neki drugi kontrolni stream. Kako to mogu napraviti?

Pomicanje miša struja bez ograničenja

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

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

Jednostavno rješenje: Korištenje ograničavanja trebala bi biti nešto poput ovoga:

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

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

- ovaj sample() operator uzeo sam s ovdje: http://reactivex.io/documentation/operators/sample.html

Ali ovo ne radi u Angular 2 - CLI projektu. Daje mi ovu grešku:

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

Još jedan snažniji način za postizanje istog rezultata, mislim da može biti sljedeći: Bilo bi sjajno ako možemo poslati najnoviji mousemove item na temelju stavki primljenih iz drugog streama. Bilo koji stream - koji smo mi stvorili..

Na primjer:

Kada primimo novu stavku (1, 2, 3..) iz eachSecond$ Streama (=naš "Control Stream"), - emitiramo dalje (u mouseMoveEachSecond$) - posljednju primljenu stavku iz mouseMove$ streama.

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}

Jednostavno je. Ovo mi je osobito bilo teško implementirati - koliko mogu zaključiti, to je zato što bi stanje toka trebalo nekako biti dijeljeno. Potreban je način da se zna što je već poslano.

U mouseMoveEachSecond$stavke dolaze jednu po jednu, redom. Ili broj ili objekt. Trebate znati:

  • Koji je bio posljednji time item kada ste primili novi mouse item.
  • I koji je bio posljednji mouse item kada ste primili novi time item.

Ovo će vam omogućiti da:

  • ne šaljite mouse item nizvodno kada miš ne pomiče na ekranu.
  • ne šaljite mouse item nizvodno dok se iz svakog eachSecond$ toka ne oslobodi novi time item.

Ovo se može napraviti korištenjem globalnih varijabli. Ali to nije RxJS način rada. Ne znam za način kako spremiti posljednji poslan predmet i koristiti ga u sljedećem koraku. Reduce operator može zadržati stanje: od jednog objavljenog predmeta do drugog, ali kako možemo koristiti to svojstvo za postizanje ovog ponašanja? Ovi streamovi moraju se smatrati beskonačnima. Hm... trebao bi postojati jednostavniji način za kombiniranje ova 2 streama... Možda je prerano... Osjećam da moj mozak ne želi razmišljati u streamovima :)

Dakle, 2 pitanja su jasno postavljena:

  1. koji su neki načini za ograničavanje streama mousemove u Angular 2?
  2. mogu li koristiti drugi tok za kontrolu kada ću otpustiti stavke mousemove? Ako da, ovaj "Kontrolni tok" može biti bilo koji tok, ili postoje ograničenja?

Hvala ti puno:)

Odgovor #1

Upravo si blizu! Ironično, tvoj čarobni operator je sample.

Stara varijanta funkcije sample (iz RxJS4) bila je preopterećena i mogla je primiti broj kao svoj argument. U RxJS5 (onoj koja dolazi s Angular2) podijeljena je tako da zapravo postoje dva operatora sample i sampleTime. Ovaj posljednji prima vremenski argument i uzima posljednji događaj u određenom vremenskom prozoru kada emitira.

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

Prvi uzima Observable kao jedini argument i emitira posljednji primljeni element svaki put kada kontrolni Observable emitira.

tj.

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));

Odgovor #2

što je s uvozom throttleTime:

import { throttleTime } from 'rxjs/operators';

i zatim ga koristiti na taj način:

mousemouve$.pipe(
      throttleTime(5000)
    )