我如何使用 Rx 处理非此即彼

Question

我需要处理反应式扩展的非此即彼类型情况。

我收到了 2 个事件（A 或 B）。 当事件 A 在 2 秒内进入后跟事件 B 时，我想生成一个 C。 如果 A 后面没有 B，那么我想生成一个 D。

有没有办法用 Rx 表达这一点？

谢谢

Answer 1

这里有一种方法可以做到这一点：

Observable.Merge(eventA, eventB)
    .Timestamp()
    .Buffer(2 /*items*/,1 /*at a time*/)
    .SelectMany(x => {
        if (x[1].Timestamp - x[0].Timestamp < TimeSpan.FromSeconds(2.0) &&
            x[0].Value.EventType == "A" && x[1].Value.EventType == "B") {
            return Observable.Return(new C());
        }
        if (x[0].Value.EventType == "A" && x[1].Value.EventType != "B") {
            return Observable.Return(new D());
        }
        return Observable.Empty<AnEvent>();
    });

Answer 2

SelectMany 以及返回备用可观察量的超时重载应该能够执行您所描述的操作。

Dim obsA = Observable.Interval(TimeSpan.FromSeconds(1)) 'whatever your source is
Dim obsB = Observable.Interval(TimeSpan.FromSeconds(0.5)) 'whatever your source is
'replace with your actual C/D generation functions
Dim final = From a In obsA
            From b In obsB.Select(Function(i) "C" & a.ToString()) _
                          .Timeout(TimeSpan.FromSeconds(2),
                                   Observable.Return("D" & a.ToString())) _
                          .Take(1)
            Select b

Answer 3

通过使用内置的Amb运算符，这似乎相当容易。

首先定义obsC和obsD如下：

var obsC = Observable.Return('C');
var obsD = Observable.Delay(
    Observable.Return('D'),
    TimeSpan.FromSeconds(2.0));

现在我想要一个"B"然后"C"可观察。我只想要在最新的"B"之后有一个"C"。如果出现两个"B"，我只想要最新的一个。在这里：

var obsBC =
    obsB
        .Select(_ => obsC)
        .Switch()
        .Take(1);

现在是将其与Amb运算符组合在一起的最后一个查询：

var query =
    from a in obsA
    from cd in Observable.Amb(obsBC, obsD)
    select cd;

我的测试表明此查询有效。但与所有这些事情一样，如果两个"A"在"B"之前出现，则不清楚该怎么办。我可以得到"D"/"C"或"C"/"C"或只是"D"或"C"吗？根据所需的行为，我的查询可能需要更改。