Rxjs - 订阅相互依存的观察对象

Question

我正在学习angular 2和rxjs。我有3个变量，AB C.

• B依赖于一个
• C的值取决于A的值与B

我想设置了观测这样的：当A更新时，B和C将自动更新。当B被更新时，C将被自动更新。我尝试了两种设置，但不能令人满意。

• 第一次设置：B订阅可观察的A; C订阅了可观察到的B ，其中最高为 A. A中的变化确实级联到B然后级联到C，但A的值不是最新的。
• 第二设置：B订阅可观察的A; ç订阅combineLatest观察到A和B.此设置的作品，但我通过B获得两个更新为C，第一，然后从A

我如何设置我的观测/订阅这样的：当A是更新后，C只会用A和B的最新值更新一次？

编辑 - 码ADDED

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)。它会迫使里面的时候withLatestFrom值已经更新C -subscription异步运行。

欲了解更多信息，请访问RxJs的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当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

好了，所以我与您的代码进行测试。调整以改变一些事情。你可以在这里看到代码：https://jsbin.com/zonedirogi/edit?html,js,console,output。主要的变化实际上是从A而不是B，最后从B变成A。然后它按预期工作。

仅供参考，您不需要使用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);