向 Observable 添加数据答案

【问题标题】：Add data to Observable向 Observable 添加数据
【发布时间】：2016-03-27 08:40:11
【问题描述】：

我有三个 Observable，它们包含从服务器获取的数据（都是相同的类型）。为了很好地显示数据，我想将空数据点添加到流中，以便时间对齐（记录数据的时间）。像这样的：

Stream 1: 12:30 ------ 15:30 -- 16:00 ----------------- 19:00
Stream 2: 12:30 --------------- 16:00 ------ 17:30 ----------
Stream 3: -------------15:30 -------------------------- 19:00
                    |                            |
                    V                            V
Stream 1: 12:30 ------ 15:30 -- 16:00 ------(17:30)---- 19:00
Stream 2: 12:30 ------(15:30)-- 16:00 ------ 17:30 ----(19:00)
Stream 3:(12:30)-------15:30 --(16:00)------(17:30)---- 19:00

括号表示为空。有没有好的方法来做到这一点？还是我必须彻底改变它？

【问题讨论】：

也许你需要像mergeconcat这样的东西？

标签： rxjs rxjs5

【解决方案1】：

如果您需要将所有 3 个流合并为 1 个，combineLatest 在这里会很有帮助。如果您需要 3 个输出流，请使用 withLatestFrom 或 sample。 withLatestFrom 允许你编写一个选择器函数，将两者结合起来，sample 直接输出原始值。请注意，与示例相比，您必须反转 withLatestFrom 上的参数。

您可能还希望当 2 个源非常接近地连续发出时，新流只发出 1 个值。我们可以通过debounce 做到这一点。

combineLatest 的代码非常简单：

Rx.Observable.combineLatest(source1, source2, source3, (s1, s2, s3) => ...)
  .debounce(10)

结果：

Stream 1: 12:30 ------ 15:30 -- 16:00 ----------------- 19:00
Stream 2: 12:30 --------------- 16:00 ------ 17:30 ----------
Stream 3: -------------15:30 -------------------------- 19:00
                    |                            |
                    V                            V
Stream:   12:30 ------ 15:30 -- 16:00 -------17:30----- 19:00

如果我们想要 3 个流，我们使用 withLatestFrom:

var streams = Rx.Observable.merge(source1, source2, source3);
s1 = source1.sample(sources).debounce(10);
s2 = source2.sample(sources).debounce(10);
s3 = stream3.sample(sources).debounce(10);

或者如果我们使用withLatestFrom:

s1 = streams.withLatestFrom(source1).debounce(10);

结果：

Stream 1: 12:30 ------ 15:30 -- 16:00 ----------------- 19:00
Stream 2: 12:30 --------------- 16:00 ------ 17:30 ----------
Stream 3: -------------15:30 -------------------------- 19:00
                    |                            |
                    V                            V
Stream 1: 12:30 ------ 15:30 -- 16:00 ------ 16:00 ---- 19:00
Stream 2: 12:30 -------12:30 -- 16:00 ------ 17:30 ---- 17:30
Stream 3: -------------15:30 -- 15:30 ------ 15:30 ---- 19:00

我不是 100% 确定您所说的“括号是空的”是什么意思。你想要这样的大理石吗？

Stream 1: a ---- b ------------ c --------- d ---------
Stream 2: g ---- i -----k------ l --------- m ---------
Stream 3: h ---- j ------------ i ----u---- a ---------
                    |                            |
                    V                            V
Stream 1: a ---- b -----_------ c ----_---- d ---------
Stream 2: g ---- i -----k------ l ----_---- m ---------
Stream 3: h ---- j -----_------ i ----u---- a ---------

_ 是一个空元素？

source1.sample(Rx.Observable.merge(source1, source2, source3))
  .debounce(10)
  .startWith("_")
  .publish(xs => xs.zip(xs.skip(1)), (a, b) => a === b ? "_", b)

然后我们可以使用sample，但过滤重复值并将它们替换为空元素。为此，我们使用了一个小技巧，比较最后两个元素，如果它们相等，我们插入一个空值。

【讨论】：

基本上我的意思是你最后一个马贝图显示的内容，但有日期。日期是模型的一部分。空模型应该包含日期和一些默认值，而不是下划线。我的函数看起来像这样：Pastebin
我想补充一点，默认值是通过一个函数传递的，该函数接受一个键并输出一个默认值。在我的示例中，键是时间码，生成的默认值是具有所需默认值和日期的模型。
@SuperManitu 我目前的答案有点草率，当我意识到sample 是一个不错的选择时，我将withLatestFrom 替换为sample，但我认为我在那里犯了一些错误。我稍后会清理它。
@SuperManitu 至于你的问题。我认为像Rx.Observable.merge(source1, source2, source3).scan(/* select max time */).debounce(1).withLatestFrom(CurrentSource, (time, current) => /* combine them */).publish(xs => xs.zip(xs.skip(1)), (a, b) => a.value === b.value ? "_", b) 这样的东西会起作用。
这是假设源上的值之间至少有 1 毫秒（或者为了安全起见，15 毫秒），但是如果 2 个流产生，它们会在一个毫秒的时间跨度内完成。而且我们可以通过比较值来过滤重复项，即源通常不会两次产生相同的值。（如果这是一个问题，我们仍然可以修改为源来防止）