如何在 Rx 中使用新的 BufferWithTimeOrCount 返回 IObservable<IObservable<T>>而不是 IObservable&lt;IList&lt;T&gt;&gt;

问题描述

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

On Windows Phone 7 there is a new version of the BufferWithTimeOrCount extension method for IObservable that returns a "stream of streams" instead of the previous "stream of lists". I'm having difficulty trying to use either the new or old methods, so maybe I just don't understand how it works, but my goal is to create a stream that only fires when an existing stream matches a specified time based pattern during the previous 2 touch events. So far I have created streams for TouchUp and TouchDown (see related question) and In pseudo code I want something like:

//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

当使用内置在 ROM 中的稳定"Microsoft.Phone.Reactive 版本的 Rx 时，调用 IObservable.BufferWithTimeOrCount(...) 返回一个 IObservable，使用标准列表操作符很容易使用(如上所述)，但由于某种原因 BufferLast2 总是返回两个 down 事件而不是 Down->我预期的顺序.

When using the "Stable" Microsoft.Phone.Reactive version of Rx that is built into the ROM calling IObservable<Class>.BufferWithTimeOrCount(...) returns a IObservable<IList<Class>>, which is pretty easy to work with using the standard list operators (as outlined above), but for some reason BufferLast2 was always returning two down events instead of the Down->Up sequence that I expected.

我认为这可能是代码中的一个错误，所以我尝试添加对 最新版本 Rx 并使用了 C:\Program Files (x86)\Microsoft Cloud Programmability\Reactive Extensions\v1.0.2838.0\WP7\System.Reactive.dll 其中 BufferWithTimeOrCount(...) 返回一个 IObservable.这使得像 Where x.Count == 2 或 Where x.First().P == ... 这样的简单过滤器更难编写.我实际上还没有想出如何在这个返回值上做一个简单的过滤器，比如 x.Count() == 2 而不创建一个完全独立的订阅或主题对象，这看起来太复杂了.这可能是一个简单的错误，就像我上一个问题一样(我只需要一个 Where 子句:-P)，但它真的让我发疯.有什么帮助吗?

I figured it might be a bug in the code, so I tried adding a reference to the latest version of Rx and used the Observable Extensions from C:\Program Files (x86)\Microsoft Cloud Programmability\Reactive Extensions\v1.0.2838.0\WP7\System.Reactive.dll in which BufferWithTimeOrCount(...) returns a IObservable<IObservable<Class>>. This makes simple filters like Where x.Count == 2 or Where x.First().P == ... much harder to write. I haven't actually figured out how to do a simple filter like x.Count() == 2 on this return value without creating a completely separate subscription or Subject object, which seams way too complex. It's probably a simple error like my last question (all I needed was a Where clause :-P) but it is really driving me bonkers. Any help?

推荐答案

更改 api 使缓冲看起来更像 Rx-y 并且适合他们的 Window 运算符实现(如果使用反射器，您将能够请参阅使用 Window 的缓冲区运算符).我认为他们改变它的原因可能有多种.我不会再猜测他们，因为他们比我聪明多了！

Changing the api makes the buffering look more Rx-y and fits with their Window operator implementation (wouldn't be surprised if using reflector you'd be able to see the Buffer operators using Window). I would think there's probably a variety of reasons that they've changed it. I'm not going to second guess them as they're a lot smarter than me!

所以这是我对解决方案的尝试.可能有一种更简洁的方法来获得您想要的东西，但我可能会实现我自己的扩展方法来缓冲到列表中.也许是这样的:

So here's my stab at a solution. There may be a cleaner way to get what you're after but i'd probably implement my own extention method to buffer into a list. Maybe something like:

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 的原因，请在 msdn 上的 rx 论坛

I suggest if you want a full explanation of why they've changed the api, go and ask the question over on the rx forums on msdn

这篇关于如何在 Rx 中使用新的 BufferWithTimeOrCount 返回 IObservable<IObservable<T>>而不是 IObservable&lt;IList&lt;T&gt;&gt;的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！