system.reactive - 我该如何处理 Rx 或 Rx？

Question

我需要使用 Reactive-Extensions 处理非此即彼的类型情况。

我收到了 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 以及返回备用可观察对象的 Timeout 重载应该能够完成您所描述的操作。

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”吗？根据所需的行为，我的查询可能需要更改。