RxJS5 组通过从第一组中删除第一项？答案

【问题标题】：RxJS5 groupBy dropping the first item from first group?RxJS5 组通过从第一组中删除第一项？
【发布时间】：2018-02-12 04:14:29
【问题描述】：

我很难理解为什么这段代码会删除第一个元素？

const Rx = require("@reactivex/rxjs")

const requestQueue = Rx.Observable.from([
  {type: "show.json", value: "1"},
  {type: "tweets.json", value: "A"},
  {type: "show.json", value: "2"},
  {type: "tweets.json", value: "B"}
])

requestQueue.groupBy(request => request.type).bufferCount(2).mergeMap(requestSubstreams => {
  return requestSubstreams[0].merge(requestSubstreams[1])
}).subscribe(x => console.log(x))

我得到的输出是：

{ type: 'tweets.json', value: 'A' }
{ type: 'show.json', value: '2' }
{ type: 'tweets.json', value: 'B' }

做了一些实验，它总是第一个被丢弃的元素。

这是 RxFiddle 的链接：http://rxfiddle.net/#type=editor&code=Y29uc3QgcmVxdWVzdFF1ZXVlID0gUnguT2JzZXJ2YWJsZS5mcm9tKFsKICB7dHlwZTogInNob3cuanNvbiIsIHZhbHVlOiAiMSJ9LAogIHt0eXBlOiAidHdlZXRzLmpzb24iLCB2YWx1ZTogIkEifSwKICB7dHlwZTogInNob3cuanNvbiIsIHZhbHVlOiAiMiJ9LAogIHt0eXBlOiAidHdlZXRzLmpzb24iLCB2YWx1ZTogIkIifQpdKQoKcmVxdWVzdFF1ZXVlLmdyb3VwQnkocmVxdWVzdCA9PiByZXF1ZXN0LnR5cGUpLmJ1ZmZlckNvdW50KDIpLm1lcmdlTWFwKHJlcXVlc3RTdWJzdHJlYW1zID0+IHsKICByZXR1cm4gcmVxdWVzdFN1YnN0cmVhbXNbMF0ubWVyZ2UocmVxdWVzdFN1YnN0cmVhbXNbMV0pCn0pLnN1YnNjcmliZSh4ID0+IGNvbnNvbGUubG9nKHgpKQ== ...但我还没有弄清楚如何解释它。

我必须求助于这段代码（使用分区）才能得到我期望的结果：

const requestQueue = Rx.Observable.from([
  {type: "show.json", value: "1"},
  {type: "tweets.json", value: "A"},
  {type: "show.json", value: "2"},
  {type: "tweets.json", value: "B"}
])

const requestSubstreams = requestQueue.partition(request => request.type == "show.json")
const showJsonSubstream = requestSubstreams[0]
const tweetsJsonSubstream = requestSubstreams[1]

showJsonSubstream.merge(tweetsJsonSubstream).subscribe(x => console.log(x))

您能否帮助指出 groupBy 的问题/警告？

【问题讨论】：

在 StackOverflow 中找到相关帖子：stackoverflow.com/questions/38043728/…
为什么groupBy 和merge 之间有一个bufferCount 运算符？您希望这会产生什么行为？
对，我注意到 bufferCount 似乎是这里的罪魁祸首（不知道如何/为什么）。但是我需要它，因为我需要获取所有分组（我想在该 mergeMap 中以某种方式连接/组合第 1 组的可观察对象和第 2 组的可观察对象）。有什么选择吗？
不清楚为什么您使用groupBy 或partition 将值分开type 只是为了稍后将它们重新组合到同一个流中。也许你应该解释你想要达到的目标是什么。如果可能，请使用大理石图。
stackoverflow.com/a/48740280/3443096

标签： reactive-programming rxjs5

【解决方案1】：

致@cartant：它与这段代码（在这段代码中）有关，字母和数字来自不同的可观察对象。

const letterArr = ["A", "B", "C", "D", "E", "F", "G", "H", "I"]
const letterStream = Rx.Observable.zip(
  Rx.Observable.from(letterArr),
  Rx.Observable.range(0, letterArr.length - 1), 
  (letter, index) => Rx.Observable.of({letter, index}).delay(
    index == 0 ? 0 : index < 5 ? (index == 2 ? 2000 : 5000) : 20000000
  )
).concatAll().multicast(new Rx.Subject())
//A emitted at 0s, B at 5s, C at 7s, D at 12s, E at 17s, F at ... never

const numberStream = Rx.Observable.zip(
  Rx.Observable.range(1, 20),
  Rx.Observable.interval(2000).startWith(-1).map(x => x +1),
  (number, index) => ({number, index})
).multicast(new Rx.Subject())
//1 emitted at 0s, 2 at 2s, 3 at 4s, ...

const slowedDownNumberStream = Rx.Observable.zip(
  numberStream,
  Rx.Observable.zip(
    letterStream,
    Rx.Observable.interval(10000).startWith(-1).map(x => x +1)
  ).timeoutWith(11000, Rx.Observable.interval(2000).startWith(-1)),
  (number, index) => ({number, index})
)

Rx.Observable.merge(letterStream, slowedDownNumberStream).subscribe(x => console.log(new Date(), " > ", x))
letterStream.connect()
numberStream.connect()

RxFiddle：http://rxfiddle.net/#type=editor&code=Y29uc3QgbGV0dGVyQXJyID0gWyJBIiwgIkIiLCAiQyIsICJEIiwgIkUiLCAiRiIsICJHIiwgIkgiLCAiSSJdCmNvbnN0IGxldHRlclN0cmVhbSA9IFJ4Lk9ic2VydmFibGUuemlwKAogIFJ4Lk9ic2VydmFibGUuZnJvbShsZXR0ZXJBcnIpLAogIFJ4Lk9ic2VydmFibGUucmFuZ2UoMCwgbGV0dGVyQXJyLmxlbmd0aCAtIDEpLCAKICAobGV0dGVyLCBpbmRleCkgPT4gUnguT2JzZXJ2YWJsZS5vZih7bGV0dGVyLCBpbmRleH0pLmRlbGF5KAogICAgaW5kZXggPT0gMCA/IDAgOiBpbmRleCA8IDUgPyAoaW5kZXggPT0gMiA/IDIwMDAgOiA1MDAwKSA6IDIwMDAwMDAwCiAgKQopLmNvbmNhdEFsbCgpLm11bHRpY2FzdChuZXcgUnguU3ViamVjdCgpKQovL0EgZW1pdHRlZCBhdCAwcywgQiBhdCA1cywgQyBhdCA3cywgRCBhdCAxMnMsIEUgYXQgMTdzLCBGIGF0IC4uLiBuZXZlcgoKY29uc3QgbnVtYmVyU3RyZWFtID0gUnguT2JzZXJ2YWJsZS56aXAoCiAgUnguT2JzZXJ2YWJsZS5yYW5nZSgxLCAyMCksCiAgUnguT2JzZXJ2YWJsZS5pbnRlcnZhbCgyMDAwKS5zdGFydFdpdGgoLTEpLm1hcCh4ID0+IHggKzEpLAogIChudW1iZXIsIGluZGV4KSA9PiAoe251bWJlciwgaW5kZXh9KQopLm11bHRpY2FzdChuZXcgUnguU3ViamVjdCgpKQovLzEgZW1pdHRlZCBhdCAwcywgMiBhdCAycywgMyBhdCA0cywgLi4uCgpjb25zdCBzbG93ZWREb3duTnVtYmVyU3RyZWFtID0gUnguT2JzZXJ2YWJsZS56aXAoCiAgbnVtYmVyU3RyZWFtLAogIFJ4Lk9ic2VydmFibGUuemlwKAogICAgbGV0dGVyU3RyZWFtLAogICAgUnguT2JzZXJ2YWJsZS5pbnRlcnZhbCgxMDAwMCkuc3RhcnRXaXRoKC0xKS5tYXAoeCA9PiB4ICsxKQogICkudGltZW91dFdpdGgoMTEwMDAsIFJ4Lk9ic2VydmFibGUuaW50ZXJ2YWwoMjAwMCkuc3RhcnRXaXRoKC0xKSksCiAgKG51bWJlciwgaW5kZXgpID0+ICh7bnVtYmVyLCBpbmRleH0pCikKClJ4Lk9ic2VydmFibGUubWVyZ2UobGV0dGVyU3RyZWFtLCBzbG93ZWREb3duTnVtYmVyU3RyZWFtKS5zdWJzY3JpYmUoeCA9PiBjb25zb2xlLmxvZyhuZXcgRGF0ZSgpLCAiID4gIiwgeCkpCmxldHRlclN0cmVhbS5jb25uZWN0KCkKbnVtYmVyU3RyZWFtLmNvbm5lY3QoKQ==

你看，我在 slowedDownNumberStream 中使用了 letterStream。

我现在的情况，我有一个单一的流作为输入（字母和数字混合在一起），所以我需要使用 groupBy... 对它们进行分区，并且因为我需要在 mergeMap 中，所以我将 bufferCount 放在groupBy 和 mergeMap。

【讨论】：