javascript - 使用 #groupBy 创建的 RxJS #zip 组

Question

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

当运行下面的代码时，实际上只有子可观察组在 #zip 中发出值——这是为什么？

https://jsbin.com/coqeqaxoci/edit?js,console

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 发出的组是 hot并且对父组 (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 和分组依据与订阅时刻的行为

   • groupBy 按预期在父级和子级中创建可观察对象

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

   • Zip 将等待您作为参数传递的每个来源中的一个值。在您的情况下，这意味着您将订阅按可观察对象分组的子项和父项，当它们都已发布时。在日志中，只有在匹配 “Check : Parent groups = 1, Child groups = 1” 上的数字后，您才会看到 "Parent group observable emits : P1"。

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

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

   • 您将在按可观察对象分组的父对象中看到所有值，因为您在它们创建后立即订阅了它们。

2. 连接并发布

   • 我对连接和发布没有完全清楚的了解，但由于您的 child 将父级作为源，因此您不需要延迟连接到它。如果您连接到父级，子级将自动开始发出其值。因此我修改了您的代码。

   • 这应该可以回答您的直接问题，但不能回答您笛卡尔积的最初目标。或许您应该将其表述为一个问题，看看人们能给出什么答案。