使用反应性扩展的心跳模式

Question

给定一个简单的方案：

a和b在一个房间里，与B进行对话B。

在谈话时，a提供了bservable talk b随后订阅talk.subscribe（string => process a say a说）。b可以同时订阅observable.interval heartbeat 作为心跳检查。

我的问题是，我应该使用哪个操作员合并/组合两个iObservable，以便在 talk 中没有 heartbeat 的项目中，b会假设A被绑架了。

请注意，我想避免一个可存储状态的变量，因为如果我不正确同步该变量，可能会导致副作用。

谢谢，

Answer 1

想象一下您要采取的状态变量，状态代表自上次辐条以来的心跳数。看起来像这样：

var stateObservable = Observable.Merge(                     //State represent number of heartbeats since A last spoke
    aSource.Select(_ => new Func<int, int>(i => 0)),        //When a talks, set state to 0
    bHeartbeat.Select(_ => new Func<int, int>(i => i + 1))  //when b heartbeats, increment state
)
    .Scan(0, (state, func) => func(state));

我们表示语言的事件作为函数将状态重置为0，而B心跳的事件则表示为增加状态。然后，我们使用Scan函数累积。

其余的现在很容易：

var isKidnapped = stateObservable
    .Where(state => state >= 2)
    .Take(1);
isKidnapped.Subscribe(_ => Console.WriteLine("A is kidnapped"));

---

编辑：

这是 n a来源的示例：

var aSources = new Subject<Tuple<string, Subject<string>>>();
var bHeartbeat = Observable.Interval(TimeSpan.FromSeconds(1)).Publish().RefCount();
var stateObservable = aSources.SelectMany(t =>
        Observable.Merge(
            t.Item2.Select(_ => new Func<int, int>(i => 0)),
            bHeartbeat.Select(_ => new Func<int, int>(i => i + 1))
        )
        .Scan(0, (state, func) => func(state))
        .Where(state => state >= 2)
        .Take(1)
        .Select(_ => t.Item1)
    );
stateObservable.Subscribe(s => Console.WriteLine($"{s} is kidnapped"));
aSources
    .SelectMany(t => t.Item2.Select(s => Tuple.Create(t.Item1, s)))
    .Subscribe(t => Console.WriteLine($"{t.Item1} says '{t.Item2}'"));
bHeartbeat.Subscribe(_ => Console.WriteLine("**Heartbeat**"));
var a = new Subject<string>();
var c = new Subject<string>();
var d = new Subject<string>();
var e = new Subject<string>();
var f = new Subject<string>();
aSources.OnNext(Tuple.Create("A", a));
aSources.OnNext(Tuple.Create("C", c));
aSources.OnNext(Tuple.Create("D", d));
aSources.OnNext(Tuple.Create("E", e));
aSources.OnNext(Tuple.Create("F", f));
a.OnNext("Hello");
c.OnNext("My name is C");
d.OnNext("D is for Dog");
await Task.Delay(TimeSpan.FromMilliseconds(1200));
e.OnNext("Easy-E here");
a.OnNext("A is for Apple");
await Task.Delay(TimeSpan.FromMilliseconds(2200));