Modifique debounceTime en los datos de transmisión: rxjs, rxjs5

Una pregunta tonta que me ha dejado perplejo. Quiero dar un tiempo de rebote diferente basado en los datos en el flujo. Yo tengo:

const fakeData = [{number: 1}, {number: 2}, {number: 3}];
const stream$ = Rx.Observable.from(fakeData);
const delayedStream$ = stream$.concatMap(x => Rx.Observable.of(x).delay(300));

delayedStream$
.concatMap(x => x.number >=2
? Rx.Observable.of(x).debounceTime(500)
: Rx.Observable.of(x).debounceTime(1000)
)
.subscribe(x => console.log(x));
// expected output: 3
// actual output:  1 ... 2 ... 3 |

http://jsbin.com/dafaxoraca/edit?js,console

El código anterior simplemente devuelve x sin unarebotar Pero si sustituyo debounceTime con retraso, el retraso funciona como se esperaba. Obviamente me estoy perdiendo algo fundamental entre los dos operadores. He revisado los documentos y no lo estoy obteniendo.

¡Gracias por tu ayuda!

Respuestas

No puedo probar esto con su caso de uso real pero usted no está usando debounceTime correctamente.

Operador debounceTime aplica rebote solo en este flujo observable y sus datos. Ya que estas encadenando concatMap y el Observable devuelto con debounceTime, el Concat siempre esperará hasta que se complete el Observable. Así que esto siempre devuelve los tres valores.

Puedes usar debounce() que espera obtener un Observable que le permita usar la demora emitiendo valores en lugar de tiempo codificado.

const fakeData = [{number: 1}, {number: 2}, {number: 3}];
const stream$ = Rx.Observable.from(fakeData);
const delayedStream$ = stream$.concatMap(x => Rx.Observable.of(x).delay(300));

delayedStream$
.debounce(val => Rx.Observable.of(true).delay(val >= 2 ? 500 : 1000))
.subscribe(x => console.log(x.number));

Ver demostración en vivo: http://jsbin.com/tifajodogi/1/edit?js,console

Esto solo emite: 3