¿Cómo limitar un evento mousemove en RxJs al usar Angular 2 + Typescript?

Estoy en el proceso de aprender RxJs. Uso Angular2 rc3. El siguiente stream funciona, pero me da demasiados eventos de movimiento del ratón. Quiero ralentizarlos usando tiempo (throttling) u otro stream de control. ¿Cómo puedo hacer eso?

Transmisión de movimiento del ratón sin limitación

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

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

Solución simple: Usar limitación debería ser algo así:

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

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

- este operador sample() lo he tomado de aquí: http://reactivex.io/documentation/operators/sample.html

Pero esto no funciona en un proyecto de Angular 2 - CLI. Me da este error:

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

Otra forma más poderosa de lograr el mismo resultado, creo que puede ser la siguiente: Sería genial si pudiéramos enviar el último mousemove item basado en los elementos recibidos de otra secuencia. Cualquier secuencia - creada por nosotros..

Por ejemplo:

Cuando recibimos un nuevo elemento (1, 2, 3..) del flujo eachSecond$ (=nuestro "Flujo de Control"), - emitimos hacia abajo (en el mouseMoveEachSecond$) - el último elemento recibido del flujo 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}

Suena simple. Encontré esto particularmente difícil de implementar; por lo que puedo decir, esto se debe a que el estado del flujo debería compartirse de alguna manera. Se necesita una forma de indicar lo que ya se ha enviado.

En el mouseMoveEachSecond$los elementos vienen uno por uno, en secuencia. O un número o un objeto. Necesitas saber:

• ¿Cuál fue el último elemento de tiempo cuando recibiste un nuevo elemento de ratón?
• ¿Y cuál fue el último elemento de ratón cuando recibiste un nuevo elemento de tiempo?

Esto te permitirá:

• no enviar un elemento de ratón corriente abajo cuando el ratón no se mueve en la pantalla.
• no enviar un mouse item corriente abajo mientras no se haya liberado un nuevo time item del flujo eachSecond$.

Esto se puede hacer usando variables globales. Pero no es la manera de hacer las cosas en RxJS. No conozco una manera de almacenar el último elemento enviado y usarlo en el siguiente paso. El operador reduce puede mantener el estado: de un elemento emitido a otro, pero ¿cómo podemos usar esa propiedad para lograr este comportamiento? Estos flujos deben considerarse infinitos. Hmm... debería haber una manera más sencilla de combinar estos dos flujos... Tal vez sea demasiado pronto... Siento que mi cerebro no quiere pensar en flujos :)

Así que las 2 preguntas quedaron claras:

1. ¿Cuáles son algunas formas de limitar un stream de mousemove en Angular 2?
2. ¿Puedo usar otro stream para controlar cuándo libero los elementos de mousemove? Si es así, este "Stream de Control" puede ser cualquier stream, o hay limitaciones?

Muchas gracias:)

Respuesta #1

¡Estás tan cerca! Irónicamente, tu operador mágico es sample.

La antigua variante de sample (de RxJS4) estaba sobrecargada y podía tomar un número como su argumento. En RxJS5 (la que se incluye con Angular2) se ha dividido de manera que en realidad hay dos operadores sample y sampleTime. Este último es el que toma un argumento de tiempo y toma el último evento en una ventana de tiempo dada cuando se emite.

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

El primero toma un Observable como su único argumento y emite el último elemento recibido cada vez que el Observable de control emite.

es decir

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

Respuesta #2

qué tal importar throttleTime:

import { throttleTime } from 'rxjs/operators';

y luego úsalo así:

mousemouve$.pipe(
      throttleTime(5000)
    )