Rxjs - Подписывание взаимозависимым наблюдаемым

Question

Я изучаю угловые 2 и rxjs. У меня есть 3 переменных, AB C.

• B зависит от значения A
• C зависит от значения A и B

Я пытаюсь настроить до наблюдаемых такие, что: Когда A обновляется, B и C будут автоматически обновляться. Когда B будет обновлен, C будет автоматически обновляться. Я попробовал две установки, но они не являются удовлетворительными.

• Первая установка: B подписывается наблюдаемым A; C подписывается наблюдаемым B withlatestfrom A. Изменения в A сделали каскадом вниз до B, затем до C, но значение от A не является последним.
• Вторая установка: B подписывается наблюдаемым A; C выписывает combineLatest наблюдаемых А и В. Этой установки работает, но я получаю два обновления для C, первый из B, а затем из А.

Как я могу настроить свои наблюдаемые/подписка, например, что, когда А обновлено, C будет обновляться только один раз с последним значением от A и B?

EDIT - КОДЫ ДОБАВЛЕНО

var A = new Rx.BehaviorSubject(1); 
 
var A_Observable = A.asObservable(); 
 
var B = new Rx.BehaviorSubject(10); 
 
var B_Observable = B.asObservable(); 
 

 
A_Observable.subscribe(function(A_value) { 
 
    var newB = A_value * 10; 
 
    // console.log("B auto updating to " + newB); 
 
    B.next(newB); 
 
}); 
 

 
// LATEST FROM OBSERVABLE 
 
var latestFromObservable = B_Observable.withLatestFrom(A_Observable); 
 
latestFromObservable.subscribe(function(data) { 
 
    console.log("LATEST FROM : Value for A is " + data[1] + " ; Value for B is " + data[0]); 
 
}); 
 

 
// COMBINE ALL OBSERVABLE 
 
var combineAllObservable = Rx.Observable.combineLatest(A_Observable,B_Observable); 
 
combineAllObservable.subscribe(function(data) { 
 
    console.log("COMBINE LATEST : Value for A is " + data[0] + " ; Value for B is " + data[1]); 
 
}); 
 

 
// UPDATE TO A 
 
setTimeout(function(){ 
 
    console.log("UPDATING A"); 
 
    A.next(2); 
 
},1000); 
 

 
// SATISFACTORY RESULT 
 
setTimeout(function(){ 
 
    console.log("SATISFACTORY RESULT : Value for A is 2 ; Value for B is 20 --- CALLED ONLY ONCE WITH LATEST VALUES"); 
 
},2000);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.1/Rx.min.js"></script>

Answer 1

код ведет себя так, потому что по умолчанию RxJs запускает код синхронно. Цепь событий для withLatestFrom случая:

1. Нажимает новое A значения: A.next(2);
2. RxjS видит, что есть два Обычаев A:
   1. B -subscription (приходит первым)
   2. withLatestFrom (на втором месте)
3. RxJs звонки B -subscription
   1. Это толкает новое значение B
   2. RxjS видит, что есть использование B: (! Со старым значением A) C -subscription
   3. RxJs называет C -subscription
4. Новое значение A вытащено в withLatestFrom, но C -подписка не активирована (поскольку она активируется только по B обновлениям)

Возможное решение добавить debounceTime(0) .Это заставит C -subscription запустить асинхронно, когда значение внутри withLatestFrom уже обновлен.

Для получения дополнительной информации google для RxJs's Scheduler.

var A = new Rx.BehaviorSubject(1); 
 
var A_Observable = A.asObservable(); 
 
var B = new Rx.BehaviorSubject(10); 
 
var B_Observable = B.asObservable(); 
 

 
A_Observable.subscribe(function(A_value) { 
 
    var newB = A_value * 10; 
 
    // console.log("B auto updating to " + newB); 
 
    B.next(newB); 
 
}); 
 

 
// LATEST FROM OBSERVABLE 
 
var latestFromObservable = B_Observable.debounceTime(0).withLatestFrom(A_Observable); 
 
latestFromObservable.subscribe(function(data) { 
 
    console.log("LATEST FROM : Value for A is " + data[1] + " ; Value for B is " + data[0]); 
 
}); 
 

 
// COMBINE ALL OBSERVABLE 
 
var combineAllObservable = Rx.Observable.combineLatest(A_Observable,B_Observable).debounceTime(0); 
 
combineAllObservable.subscribe(function(data) { 
 
    console.log("COMBINE LATEST : Value for A is " + data[0] + " ; Value for B is " + data[1]); 
 
}); 
 

 
// UPDATE TO A 
 
setTimeout(function(){ 
 
    console.log("UPDATING A"); 
 
    A.next(2); 
 
},1000); 
 

 
// SATISFACTORY RESULT 
 
setTimeout(function(){ 
 
    console.log("SATISFACTORY RESULT : Value for A is 2 ; Value for B is 20 --- CALLED ONLY ONCE WITH LATEST VALUES"); 
 
},2000);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.1/Rx.min.js"></script>

Я хочу добавить две вещи:

1. withLatestFrom подход лучше в вашем случае, поскольку зависимость график выглядит C -> B -> A и когда обновляется B обновляется тоже. Если бы A и B были независимыми, то combineLatest был бы лучшим выбором.
2. Нет необходимости в Subject для B. Вы можете переписать код, как это (не debounceTime!)

var A = new Rx.BehaviorSubject(1); 
 
var A_Observable = A.asObservable(); 
 

 
var B_Observable = A_Observable.map(function(A_value) { 
 
    var newB = A_value * 10; 
 
    // console.log("B auto updating to " + newB); 
 
    return newB; 
 
}); 
 

 
var latestFromObservable = B_Observable.withLatestFrom(A_Observable); 
 
latestFromObservable.subscribe(function(data) { 
 
    console.log("LATEST FROM : Value for A is " + data[1] + " ; Value for B is " + data[0]); 
 
}); 
 

 
// UPDATE TO A 
 
setTimeout(function(){ 
 
    console.log("UPDATING A"); 
 
    A.next(2); 
 
},1000);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.1/Rx.min.js"></script>

Answer 2

Alright, поэтому я тестировал с вашим кодом. Измените несколько вещей. Вы можете увидеть код здесь: https://jsbin.com/zonedirogi/edit?html,js,console,output. Основное изменение на самом деле вместо B с последним от A, вместо A заменяется на A с последней из B. Затем он работает так, как ожидалось.

FYI, вам не нужно использовать asObservable() для преобразования, подлежащего наблюдению.

var a = new Rx.BehaviorSubject(1); 
var b = new Rx.BehaviorSubject(10); 

a.subscribe(x => { 
    b.next(x * 10); 
}); 

a.withLatestFrom(b).subscribe(x => { 
    console.log(x) 
}) 

// UPDATE TO A 
setTimeout(function(){ 
    console.log("UPDATING A"); 
    a.next(2); 
}, 1000);