Другие операторы в цепочке вычислений, чем combLatest, чтобы избежать избыточных вычислений

Question

Я успешно перенесла вычисление Excel в JS с использованием RxJS. Любые входные данные представляются как Наблюдаемые, а последующие вычисления выполняются с .map и .combineLatest, когда в любой формуле Excel используется более одного ввода.

Это отлично работает, за исключением одной проблемы. Вот упрощенный пример:

Три вход (a$=1, b$=2, c$=3) используется в двух различных расчетах ($ab = $a+$b = 3 в первом, $bc = $b+$c = 5 во втором) в качестве промежуточных шагов для расчета конечного результата $abbc = $ab + $bc = 8.

Когда $ b теперь обновляется/испускает новое входное значение (например, 4), $ abbc вычисляется дважды - сначала, когда $ ab обновляется (приводя к неправильному результату $abbc=10) в правильном результате 12.

Хотя конечный результат верен, промежуточный расчет является ошибочным и избыточным. Есть ли способ только выполнить последний расчет в случае, если обновляется $b - при этом также обновляется расчет, когда обновляется a$ или c$ (что исключало бы оператор zip). Я понимаю, что этот пример, очевидно, может быть упрощен, чтобы исключить промежуточные этапы и рассчитал $abbc непосредственно с $a, $b и $c - но в реальном живом примере это невозможно/выполнимо.

Вот пример работает в JSbin: https://jsbin.com/pipiyodixa/edit?js,console

Answer 1

Проблема здесь заключается в том, что поведение RxJS является правильным по своему дизайну.

Это действительно необходимо сначала обновить b =>ab =>abbc и затем bc =>abbc. Он обрабатывал значения в потоках.

Вы хотите обработать значения в «слоях». a, b, c, затем ac, bc и после этого рассчитать окончательное значение abbc.

Единственный способ, которым я могу думать, - использовать контекст выполнения JavaScript и трюк с помощью setTimeout(() => {}, 0). Таким образом, вы не планируете какой-либо тайм-аут (на самом деле реальный timeout might be > 0) и просто запустите закрытие в другом контексте выполнения после того, как JavaScript завершит выполнение текущего.

Плохо то, что, чтобы избежать повторного излучающие значения несколько раз, нужно кэшировать даже больше (из-за merge()):

var b2$ = b$.cache(1); 

var ab$ = a$ 
    .combineLatest(b2$, (a, b) => a + b) 
    .do(x => console.log('$a + $b = ' + x)) 
    .cache(1); 

var bc$ = b2$ 
    .combineLatest(c$, (b, c) => b + c) 
    .do(x => console.log('$b + $c = ' + x)) 
    .cache(1); 

var abbc$ = new Rx.Observable.merge(ab$, bc$) 
    .auditTime(0) 
    .withLatestFrom(ab$, bc$, (_, ab, bc) => ab + bc) 
    .do(x => console.log('$ab + $bc = ' + x)); 

console.log("Initial subscription:") 
abbc$.subscribe(); 

b$.next(4); 

Оператор auditTime() самое главное здесь. Он вызывает withLatestFrom(), чтобы обновить его значение, когда он запускается в первый раз на ab$ или bc$, он игнорирует все последовательные испускания до конца этого закрытия (это трюк setTimeout()).

Смотреть демо: https://jsbin.com/zoferid/edit?js,console

Кроме того, если вы добавите a$.next(5); окончательный расчет выполняется только один раз (что может быть как хорошо или плохо :)).

Я не знаю, решает ли эта проблема, потому что, как я сказал, RxJS работает таким образом. Я не тестировал его на более сложном примере, поэтому, возможно, это не так, как вы можете использовать в конце.

Обратите внимание, что cache() оператор был удален в RC.1 и нет никакой замены на данный момент: https://github.com/ReactiveX/rxjs/blob/master/CHANGELOG.md

Answer 2

Основываясь на @ (правильный) ответ Мартина, я создал оператор rxjs, который делает combineLatestDelayed:

Rx.Observable.prototype.combineLatestDelayed = function(b$, cb) { 
    var a2$ = this.cache(1); 
    var b2$ = b$.cache(1); 
    return a2$ 
    .merge(b2$) 
    .auditTime(0) 
    .withLatestFrom(a2$, b2$, (_, a, b) => cb(a,b)); 
} 

Таким образом, мне нужно лишь заменить оригинальные .combineLatest звонки с .combineLatestDelayed:

var a$ = new Rx.BehaviorSubject(1).do(x => console.log("Emitting $a=" + x)); 
var b$ = new Rx.BehaviorSubject(2).do(x => console.log("Emitting $b=" + x)); var b1$ = b$.cache(1); 
var c$ = new Rx.BehaviorSubject(3).do(x => console.log("Emitting $c=" + x)); 

var ab$ = a$ 
    .combineLatestDelayed(b1$, (a, b) => a + b) 
    .do(x => console.log('ab$: $a + $b = ' + x)) 

var bc$ = b1$ 
    .combineLatestDelayed(c$, (b, c) => b + c) 
    .do(x => console.log('bc$: $b + $c = ' + x)) 

var abbc$ = ab$ 
    .combineLatestDelayed(bc$, (ab, bc) => ab + bc) 
    .do(x => console.log('$abbc: $ab + $bc = ' + x)); 

console.log("Initial subscription:") 
abbc$.subscribe(); 

setTimeout(() => b$.next(4), 100); 

Полный JS Bin here