Comment obtenir une valeur limitée dans rxjs?

Question

Dans l'exemple des gaz de site Web rxjs,

La séquence émise est 0 4 2 1 3.

La séquence de sortie est 0 1. (Parce 4, 2 et 1 est rayé)

var Rx = require('rxjs/Rx'); 
var times = [ 
    { value: 0, time: 100 }, 
    { value: 1, time: 600 }, 
    { value: 2, time: 400 }, 
    { value: 3, time: 900 }, 
    { value: 4, time: 200 } 
]; 

// Delay each item by time and project value; 
var source = Rx.Observable.from(times) 
    .flatMap(function (item) { 
    return Rx.Observable 
     .of(item.value) 
     .delay(item.time); 
    }) 
    .throttleTime(300 /* ms */); 

var subscription = source.subscribe(
    function (x) { 
    console.log('Next: %s', x); 
    }, 
    function (err) { 
    console.log('Error: %s', err); 
    }, 
    function() { 
    console.log('Completed'); 
    }); 

La console affichera

Next: 0 (at 100ms) // The value 4 was dropped(at 200ms) 
        // The value 2 was dropped(at 400ms) 
Next: 1 (at 600ms) 
        // The value 3 was dropped(at 900ms) 
Completed 

Mais, est-il possible d'obtenir le flux de valeur perdue?

Next: 4 (at 200ms) 
Next: 2 (at 400ms) 
Next: 3 (at 900ms) 
Completed 

Answer 1

1. index à la source Fixer.
2. Combine la dernière valeur d'index de la source et étranglée.
3. Comparer l'indice de la source et étranglé. Si index source> index réduit, la source n'est pas limitée.
4. Enlever l'index en retard.

Vous pouvez utiliser cette technique dans un autre cas NOT.

var Rx = require('rxjs/Rx'); 
var times = [ 
    { value: 0, time: 100 }, 
    { value: 1, time: 600 }, 
    { value: 2, time: 400 }, 
    { value: 3, time: 900 }, 
    { value: 4, time: 200 } 
]; 

// Delay each item by time and project value; 
var source = Rx.Observable.from(times) 
    .mergeMap(function (item) { 
     return Rx.Observable 
      .of(item.value) 
      .delay(item.time); 
    }); 

var indexedSource = source 
    .scan((_, value, index) => { 
     // console.log(`value = ${value}, index = ${index}`) 
     return [value, index]; 
    }, undefined) 
    .share(); 

var indexedThrottled = indexedSource 
    .throttleTime(300 /* ms */); 

var throttled = indexedThrottled 
    .map(value => value[0]); 

var notThrottled = Rx.Observable.combineLatest(indexedThrottled, indexedSource) 
    .filter(combined => { 
     var filteredIndex = combined[0][1]; 
     var sourceIndex = combined[1][1]; 

     return sourceIndex > filteredIndex ? true : false; 
    }) 
    .map(combined => { 
     return combined[1][0]; 
    }); 

source.subscribe(value => console.log(`source : ${value}`)); 
throttled.subscribe(value => console.log(`++++++ : ${value}`)); 
notThrottled.subscribe(value => console.log(`------ : ${value}`));