使用#groupBy创建的RxJS #zip组

Question

我需要压缩分组的可观察对象(以形成相关组的笛卡尔积，但这与问题无关)。

当运行下面的代码时，只有子可观察组在#zip中发出值——为什么呢?

https://jsbin.com/coqeqaxoci/edit?js,控制台

var parent = Rx.Observable.from([1,2,3]).publish();
var child = parent.map(x => x).publish();
var groupedParent = parent.groupBy(x => x);
var groupedChild = child.groupBy(x => x);
Rx.Observable.zip([groupedChild, groupedParent])
  .map(groups => {
    groups[0].subscribe(x => console.log('zipped child ' + x)); // -> emitting
    groups[1].subscribe(x => console.log('zipped parent ' + x)); // -> not emitting
  })
  .subscribe();
groupedChild.subscribe(group => {
  group.subscribe(value => console.log('child ' + value)); // -> emitting
});
groupedParent.subscribe(group => {
  group.subscribe(value => console.log('parent ' + value)); // -> emitting
});
child.connect();
parent.connect();

编辑:正如user3743222在回答中解释的那样，groupBy发出的组是热的，并且对父组(groups[1])的订阅发生在已经发出第一个值之后。这发生在#zip等待groupedChild和groupedParent发出时，后者发出得更快(意味着它的组在#zip函数运行之前发出值)。

Answer 1

我将你的代码修改如下:

var countChild = 0, countParent = 0;
function emits ( who ) {
  return function ( x ) {console.log(who + " emits : " + x);};
}
function checkCount ( who ) {
  return function ( ) {
    if (who === "parent") {
      countParent++;
    }
    else {
      countChild++;
    }
    console.log("Check : Parent groups = " + countParent + ", Child groups = " + countChild );
  };
}
function check ( who, where ) {
  return function ( x ) {
    console.log("Check : " + who + " : " + where + " :" + x);
  };
}
function completed ( who ) {
  return function () { console.log(who + " completed!");};
}
function zipped ( who ) {
  return function ( x ) { console.log('zipped ' + who + ' ' + x); };
}
function plus1 ( x ) {
  return x + 1;
}
function err () {
  console.log('error');
}
var parent = Rx.Observable.from([1, 2, 3, 4, 5, 6])
    .do(emits("parent"))
    .publish();
var child = parent
    .map(function ( x ) {return x;})
    .do(emits("child"))
//    .publish();
var groupedParent = parent
    .groupBy(function ( x ) { return x % 2;}, function ( x ) {return "P" + x;})
    .do(checkCount("parent"))
    .share();
var groupedChild = child
    .groupBy(function ( x ) { return x % 3;}, function (x) {return "C" + x;})
    .do(checkCount("child"))
    .share();
Rx.Observable.zip([groupedChild, groupedParent])
//    .do(function ( x ) { console.log("zip args : " + x);})
    .subscribe(function ( groups ) {
                 groups[0]
                     .do(function ( x ) { console.log("Child group observable emits : " + x);})
                     .subscribe(zipped('child'), err, completed('Child Group Observable'));
                 groups[1]
                     .do(function ( x ) { console.log("Parent group observable emits : " + x);})
                     .subscribe(zipped('parent'), err, completed('Parent Group Observable'));
               }, err, completed('zip'));
//child.connect();
parent.connect();

输出如下:

"parent emits : 1"
"child emits : 1"
"Check : Parent groups = 0, Child groups = 1"
"Check : Parent groups = 1, Child groups = 1"
"Parent group observable emits : P1"
"zipped parent P1"
"parent emits : 2"
"child emits : 2"
"Check : Parent groups = 1, Child groups = 2"
"Check : Parent groups = 2, Child groups = 2"
"Parent group observable emits : P2"
"zipped parent P2"
"parent emits : 3"
"child emits : 3"
"Check : Parent groups = 2, Child groups = 3"
"Parent group observable emits : P3"
"zipped parent P3"
"parent emits : 4"
"child emits : 4"
"Child group observable emits : C4"
"zipped child C4"
"Parent group observable emits : P4"
"zipped parent P4"
"parent emits : 5"
"child emits : 5"
"Child group observable emits : C5"
"zipped child C5"
"Parent group observable emits : P5"
"zipped parent P5"
"parent emits : 6"
"child emits : 6"
"Parent group observable emits : P6"
"zipped parent P6"
"Child Group Observable completed!"
"Child Group Observable completed!"
"Parent Group Observable completed!"
"Parent Group Observable completed!"
"zip completed!"

这里有两点需要说明:

1. zip和group的行为与订阅时刻

   • groupBy按预期在parent和child中创建可观察对象

   使用这些值，您可以在日志中检查Child创建三个组，Parent创建两个

   • Zip将等待在您作为参数传递的每个源中具有一个值。在您的示例中，这意味着您将在发布子和父观察对象时订阅它们。在日志中，只有匹配了"Check : Parent groups = 1, Child groups = 1"上的数字后，才会看到"Parent group observable emits : P1"。

   • 然后订阅两个分组的可观察对象，并记录从那里产生的任何内容。这里的问题是，父组by可观察对象有一个值要传递，但是子组by可观察对象是在此之前创建的，并且已经传递了它的值，所以当你在事后订阅时，你看不到这个值，但你会看到下一个。

   • 因此，[1-3]中的值将生成3个新的按可观察对象分组的子对象，并且您将看不到其中的任何一个，因为您订阅太晚了。但是您将在[4-6]中看到值。您可以查看日志:"zipped child C4"等

   • 你会看到所有的值都在父级的可观察对象中分组，因为你在它们创建后就立即订阅了它们。

2. 连接并发布

   • 我没有一个完全清楚的理解连接和发布，但作为你的孩子有父母作为一个来源，你不需要延迟连接到它。如果连接到父进程，则子进程将自动开始发出其值。所以我修改了你的代码

   • 这应该回答了你的直接问题，但不是你最初的目标笛卡尔产品。也许你应该把它变成一个问题，看看人们能给出什么答案。