CODEWITHSUNDEEP
system.reactive
Greg Bray
Greg Bray

Reputation: 15697

How to use the new BufferWithTimeOrCount in Rx that returns IObservable<IObservable<T>> instead of IObservable<IList<T>>

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

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.

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?

Upvotes: 2

Views: 1293

Answers (2)

Bart De Smet
Bart De Smet

Reputation: 31

The latest release of Rx, v1.0.2856.0, provides both buffers and windows. For the buffers, we restored the original signatures, based on IList. The corresponding window operators will return nested observable sequences.

The way the Buffer* operators are implemented is by composing the corresponding Window* operator with the new ToList extension method that takes an IObservable into an IObservable>. All the Buffer* operator does is invoke this new ToList operator in a SelectMany selector.

Upvotes: 3

James Hay
James Hay

Reputation: 12700

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();
               });
          });
    }
}

Then something like:

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);
}

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

Upvotes: 3

Related Questions