system.reactive - 如何在 Rx 中使用新的 BufferWithTimeOrCount 返回 IObservable> 而不是 IObservable>

Question

在 Windows Phone 7 上，IObservable 有一个新版本的 BufferWithTimeOrCount 扩展方法，它返回“流的流”而不是以前的“列表流”。我在尝试使用新方法或旧方法时遇到困难，所以也许我只是不明白它是如何工作的，但我的目标是创建一个流，仅在现有流与指定的基于时间的模式匹配时触发前 2 个触摸事件。到目前为止，我已经为 TouchUp 和 TouchDown 创建了流(请参阅 related question )，在伪代码中我想要类似的东西:

//BufferLast2 should contain the last 1 or 2 touch events that occurred in the last 500ms. If no touches occurred this should return an empty set
var BufferLast2 = TouchDown.Merge(TouchUp).BufferWithTimeOrCount(TimeSpan.FromSeconds(0.5), 2);
//Use BufferLast2 to detect tap (TouchDown then TouchUp occuring in less than 0.5s)
var TouchTap = from touch2 in BufferLast2
               where touch2.Count == 2 && touch2.First().Action == TouchAction.Down && touch2.Last().Action == TouchAction.Up
               select touch2.First(); //returns initial TouchDown event
//Use BufferLast2 to detect Hold (TouchDown with no TouchUp occuring in 0.5s)
var TouchHold = from touch2 in BufferLast2
                where touch2.Count == 1 && touch2.First().Action == TouchAction.Down
                select touch2.First(); //returns initial TouchDown event

使用“稳定”时Microsoft.Phone.Reactive ROM 中内置的 Rx 版本调用 IObservable<Class>.BufferWithTimeOrCount(...)返回 IObservable<IList<Class>> ，使用标准列表运算符(如上所述)非常容易使用，但由于某种原因 BufferLast2 总是返回两个向下事件，而不是我期望的向下->向上序列。

我认为这可能是代码中的错误，因此我尝试添加对 latest version 的引用Rx 并使用了来自 C:\Program Files (x86)\Microsoft Cloud Programmability\Reactive Extensions\v1.0.2838.0\WP7\System.Reactive.dll 的可观察扩展，其中 BufferWithTimeOrCount(...)返回 IObservable<IObservable<Class>> 。这使得简单的过滤器像 Where x.Count == 2或Where x.First().P == ...更难写。我实际上还没有弄清楚如何做一个像 x.Count() == 2 这样的简单过滤器在此返回值上，无需创建完全独立的订阅或主题对象，这看起来太复杂了。这可能是一个简单的错误，就像我的上一个问题(我所需要的只是一个Where子句:-P)，但这确实让我发疯。有什么帮助吗？

Answer 1

更改 api 使缓冲看起来更 Rx-y 并适合其 Window 运算符实现(如果使用反射器，您将能够使用 Window 看到 Buffer 运算符，则不会感到惊讶)。我认为他们改变它可能有多种原因。我不会再次猜测他们，因为他们比我聪明得多!

这是我尝试的解决方案。可能有一种更干净的方法来获取您想要的东西，但我可能会实现我自己的扩展方法来缓冲到列表中。也许是这样的:

public static class BufferToList
{
   public static IObservable<IEnumerable<TSource>> BufferToList<TSource>(this IObservable<TSource> source)
    {
       return Observable.CreateWithDisposable<IEnumerable<TSource>>(observer => 
          {
             var list = new List<TSource>();

             return source.Subscribe(list.Add,
               observer.OnError,
               () =>
               {
                  observer.OnNext(list);
                  observer.OnCompleted();
               });
          });
    }
}

然后是这样的:

TouchDown.Merge(TouchUp)
   .BufferWithTimeOrCount(TimeSpan.FromSeconds(0.5), 2)
   .Select(bufferedValues => bufferedValues.BufferToList())
   .Subscribe(OnBufferOpen)

private void OnBufferOpen(IObservable<IEnumerable<IEvent<IEventArgs>>> bufferedListAsync)
{
   bufferedListAsync.Where(list => list.Count() == 2);
}

我建议，如果您想了解他们更改 api 的原因的完整解释，请在 rx forums on msdn 上提问。