Filtering a Touch.FrameReported IObservable using arbitrary boolean condition that changes over time

Question

I've been playing around with the Reactive Extensions (RX) in Windows Phone 7 and am very close to a working solution but got caught up on one small detail. I am trying to process the raw touch events using Touch.FrameReported and Observable.FromEvent (a bit of an educational quest to learn the Touch API and RX better), but I only want to process the events under certain conditions. For example I may want to filter the subscription to the touch down and touch up events only when a specific page is selected in a pivot control, but it could be any arbitrary condition that changes back and forth between true and false. Since the condition is a value that changes over time it feels like it should be another observable that gets merged with the touch event stream, but I can't for the life of me figure out how to do that.

Instead I have a semi-working solution using the TakeWhile and SkipUntil extensions for IObservable. I have a stream that receives all the touch events for the whole app (oTouchApp), a second stream that only takes items while my filtering condition is true (oTouchPage), and then two other streams that filter the touches into touch down (oTouchDown) and touch up (oTouchDown) actions. All of these streams are of type IObservable<IEvent<TouchFrameEventArgs>>, so they can easily be merged and compared to create custom gestures. The problem is that I can't get the the oTouchPage stream to restart once the filter condition changes from true to false. I'd have to manually recreate the stream, where as I would prefer that it somehow toggles itself between on and off.

Here is the code I have so far. Any help on how to filter a stream using a boolean value (like an on/off switch) would be greatly appreciated.

var oTouchApp = Observable.FromEvent<TouchFrameEventHandler, TouchFrameEventArgs>(x => new TouchFrameEventHandler(x), ev => Touch.FrameReported += ev, ev => Touch.FrameReported -= ev);
//This stops working after TestCondition goes from True to False
var oTouchPage = oTouchApp.SkipWhile((x) => TestCondition == False).TakeWhile((x) => TestCondition == True);
var oTouchDown = from t in oTouchPage
                 let primarypoint = t.EventArgs.GetPrimaryTouchPoint(this)
                 where primarypoint.Action == TouchAction.Up
                 select t;
var oTouchUp = from t in oTouchPage
               let primarypoint = t.EventArgs.GetPrimaryTouchPoint(this)
               where primarypoint.Action == TouchAction.Down
               select t;
//Code for testing
var sub1 = oTouchPage.Subscribe(x =>{Debug.WriteLine("TouchPage");});
var sub2 = oTouchDown.Subscribe(x =>{Debug.WriteLine("TouchDown");});
var sub3 = oTouchUp.Subscribe(x =>{Debug.WriteLine("TouchUp");});

UPDATE: Turns out all I needed was to add a simple where clause to the oTouchPage stream: var oTouchPage = oTouchApp.Where((x) => TestCondition == True); It may not be the best solution, since the TestCondition is evaluated each time an item is produced, but it works well and is easy to read. If the test condition was based on events or some other condition that was easy to convert into an IObservable, then I think the Window or SelectMany approaches mentioned below might be better, but then you may have to deal with a "Stream of streams". I'm fighting that right now in a related question.

Answer 1

TakeWhile triggers completion when the predicate returns false, at which point everything is torn down.

You have two choices:

The simplest solution is to use Repeat to start the whole process again:

var oTouchPage = oTouchApp
    .SkipWhile((x) => TestCondition == false)
    .TakeWhile((x) => TestCondition == true)
    .Repeat();

The other solution would be to track the situation using SelectMany to trigger the start of each "round" of listening (this is very similar to the "drag-drop" WPF Rx Example):

// Think of Subject like an observable variable
// This can just be an IObservable<bool> if the triggers actually come from elsewhere
Subject<bool> testConditionValues = new Subject<bool>();

var startTrigger = testConditionValues
    .DistinctUntilChanged()
    .Where(v => v == true);

var endTrigger = testConditionValues
    .Where(v => v == false);

var values = from start in startTrigger
             from touch in touchEvents.TakeUntil(endTrigger)
             select touch;

// Start listening
testConditionValues.OnNext(true);

// Stop listening
testConditionValues.OnNext(false);

Answer 2

In addition to Richard Szalays answer you could also look into the Window operator (although I think his second solution is probably the right one). You have windows of time that you'd like to receive touch events.

private void SetupStream()
{
    Subject<bool> testConditionValues = new Subject<bool>();

    var startTrigger = testConditionValues.DistinctUntilChanged()
         .Where(v => v == true);

    var endTrigger = testConditionValues
         .Where(v => v == false);

    touchEvents.Window(startTrigger, _ => endTrigger)
        .Subscribe(HandleNewWindow);

    // Open window
    testConditionValues.OnNext(true);

    // Close window
    testConditionValues.OnNext(false);

}

public void HandleNewWindow(IObservable<IEvent<EventArgs>> events)
{
    events.Subscribe(mousEvent => Trace.WriteLine(mousEvent));
}