rxjs - rxjs - 在 Observables 中迭代

Question

每次生成新值时，我都需要将其与所有以前的值进行比较，并且只有在满足条件时才会将其添加到流中。

如何使用 observables 做到这一点？

Answer 1

这是一个例子，可能有点复杂，它应该做一些类似于你正在寻找的事情

import {Observable} from 'rxjs';

const values = ['aa', 'ab', 'ac', 'ba', 'db', 'bc', 'cc', 'cb', 'cc']

Observable.from(values)
// accumulate all previous values into an array of strings
.scan((previousValues, thisValue) => {
    previousValues.push(thisValue)
    return previousValues
}, [])
// create an object with the previous objects and the last one
.map(previousValues => {
    const lastValue = previousValues[previousValues.length - 1]
    return {previousValues, lastValue}
})
// filters the ones to emit based on some similarity logic
.filter(data => isNotSimilar(data.lastValue, data.previousValues))
// creates a new stream of events emitting only the values which passed the filter
.mergeMap(data => Observable.of(data.lastValue))
.subscribe(
    value => console.log(value)
)

function isNotSimilar(value: string, otherValues: Array<string>) {
    const otherValuesButNotLast = otherValues.slice(0, otherValues.length - 1);
    const aSimilar = otherValuesButNotLast.find(otherValue => otherValue[0] === value[0]);
    return aSimilar === undefined;
}