system.reactive - 在 RX 中如何组合两个不同类型的源

Question

设置：

• 首先IObservable产生类型A 的值
• 第二个IObservable产生类型B的值
• 它们以不同的速度产生价值（非常快，最多每 10 毫秒）

我想要达到的目标：

每N次（N 相当慢，大约 500 毫秒）必须对服务进行调用，并提供来自First和Second IObservable的最新值。

问题： 我想知道如何使用 RX。

当前解决方案（不工作）：

var stateObs = from drag in dragObs.MostRecent(0).ToObservable()
                from roll in rollObs.MostRecent(0).ToObservable()
                select new ClientState
                            {
                                FileDragPerc = drag,
                                PhoneRoll = roll,
                                PendingFileType = FileType.Image,
                                TransferState = TransferState.SelectiveTransfer
                            };

stateObs.Sample(TimeSpan.FromMilliseconds(300))
        .Subscribe(x => _lsService.SetClientStateAsync(x),
                    x => Debug.WriteLine("Error in observable "),
                    () => Debug.WriteLine("Error observable finished! "));

Answer 1

你是对的。这就是CombineLatest运营商所做的：

A: 1...2...3...4...5...

B: a.....b.........c...

保留任一序列的最后一个值以生成选择器作用的对。输出流将是(1,a) (2,a) (2,b) (3,b)等等。

如果您必须从任一流中构造正确的对，请使用“Zip”运算符，它将为您提供(1,a) (2,b) (3,c)等等。

附言

我建议尝试更好地理解编译器如何重写查询理解。它将解决您的大部分困惑。

from a in oA
from b in oB
select ...

有效地SelectMany(oA, oB)

Answer 2

我想我自己找到了答案

CombineLatest() 操作可以满足我的需要，这就是我得到的：

var stateObs = dragObs.CombineLatest(rollObs, (d, r) => new ClientState
                                                       {
                                                           FileDragPerc = d,
                                                           PhoneRoll = r,
                                                           TransferState = TransferState.SelectiveTransfer,
                                                           PendingFileType = FileType.Image
                                                       });

    stateObs.Sample(TimeSpan.FromMilliseconds(300))
            .Subscribe(x => _lsService.SetClientStateAsync(x),
                        x => Debug.WriteLine("Error in observable "),
                        () => Debug.WriteLine("Error observable finished! "));