0

我从 Joseph Albahari 那里得到一个观察结果:

IObservable<int> GetHeartBeat()
{
    return Observable.Create<int>((observer, cancelToken) => this.Start(observer, cancelToken));
}

async Task Start(IObserver<int> observer, CancellationToken cancelToken)
{
    int beat = 0;
    var random = new Random();
    while (beat < 10)
    {
        await Task.Delay(random.Next(500) + 700, cancelToken);
        observer.OnNext(beat);
        beat++;
    }
}

然后我运行这个:

var observable = this.GetHeartBeat()
    .TimeInterval()
    .Buffer(3, 1)
    .Select(l => l.Average(x => 60 / x.Interval.TotalSeconds));
observable
    .Select(i => string.Format("{0}, ", i))
    .Concat(observable.Count().Select(i => string.Format("{0}", i)))
    .Subscribe(i => System.Diagnostics.Debug.WriteLine(i));
observable.Wait();

我希望看到这个:

71.9889864709428, 1
72.9592264002639, 2
67.6743699542054, 3

等等

但我看到了这个:

71.9889864709428, 
72.9592264002639, 
67.6743699542054, 

我现在能错到什么程度?

4

2 回答 2

1

我认为你需要这个:

var observable = this.GetHeartBeat()
    .TimeInterval()
    .Buffer(3, 1)
    .Select((l, i) => string.Format("{0}, {1}", l.Average(x => 60 / x.Interval.TotalSeconds), i))
    .Subscribe(i => System.Diagnostics.Debug.WriteLine(i));
于 2013-11-07T23:54:17.793 回答
1

使用 Rx 获得运行总计的标准方法是使用Scan运算符。

于 2013-11-08T18:51:41.037 回答