RX fluent subscribing to observables

Question

I have three observables:

var itemsManipulationStarted =
 Observable.FromEvent<ManipulationStartedEventArgs>(Items, "ManipulationStarted");
var pageManipulationDelta = Observable.FromEvent<ManipulationDeltaEventArgs>(this, "ManipulationDelta");
var pageManipulationCompleted =
 Observable.FromEvent<ManipulationCompletedEventArgs>(this, "ManipulationCompleted");

They are always called in that sequense: itemsManipulationStarted -> pageManipulationDelta -> pageManipulationCompleted

After every call to any observable I need to make some actions:

itemsManipulationStarted.Subscribe(e => Items.IsHitTestVisible = false);
pageManipulationDelta.SkipUntil(itemsManipulationStarted)
 .Subscribe(e => e.EventArgs.Complete());
pageManipulationCompleted.SkipUntil(pageManipulationDelta)
 .Subscribe(e => Items.IsHitTestVisible = true);

Is it clear or is there another way to write some "fluent" subscribing?

• Dan: To be clear.. do you rely on these events working in this order?
• Veikedo: @DanPantry, yes

Answer 1

If you're on C# 5.0, you can use await with IObservables to write code like this:

await itemsManipulationStarted.FirstAsync();
Items.IsHitTestVisible = false;
var e = await pageManipulationDelta.FirstAsync();
e.EventArgs.Complete();
await pageManipulationCompleted.FirstAsync();
Items.IsHitTestVisible = true;

Though I'm not completely sure this will do what you want, because of timing.

• Veikedo: Actually I'm under windows phone 7 :\ Thanks!
• Veikedo: wow. Actually I can use async\await. I'll try your answer. Thanks again!
• Veikedo: If when I'm awaiting xxx.FirstAsync() and the IObservable becomes disposed, I would got an exception?
• Dan: Well, yes. The observable becomes disposed so the element you are awaiting becomes unreachable. It might be worth instead re-writing your code to not use .FirstAsync() (although you can do this) and instead utilize the various methods in Rx to combine sequences. All of this could be feasibly combined into a single sequence.

Answer 2

As per your comment, it appears that you rely on events working in this order. This seems like a bit of an anti-pattern to me to give the end-user the freedom to use all of these events, but rely on them working in the order that they do.

As a result, I would recommend removing the "ManipulationStarted" and "ManipulationCompleted" events. Instead, I would suggest something more along the lines of this:

pageManipulationDelta.Subscribe(e => e.EventArgs.Complete());

And then only fire this event - the "ManipulationDelta" event during the period of Manipulation being started or ended (whatever manipulation is). There is almost certainly a better way to approach your issue. There is no need for you to have these 3 observables 'flow' into each other. Instead, you should only expose one observable - let the event publisher decide on any security restrictions... such as state mutation, which is against the nature of Rx itself anyway.

Revised code:

var pageManipulationDelta = Observable.FromEvent<ManipulationDeltaEventArgs>(this, "ManipulationDelta");
....
// When we start manipulation
IsManipulationStarted = true;
ManipulationDelta(args);
IsManipulationStarted = false;

Although the fact there is the mutation of a boolean state variable confuses me - why is this necessary? To give a more indepth review of your code I will need to see the reason why you have approached it the way you have.

Note that this is not tested:

 var combinedObservable = Observable.Defer(() =>
 {
 var allowPublications = false;
 var subscription = startObservable.Select(_ => true)
 .Merge(endObservable.Select(_ => false))
 .Throttle(TimeSpan.FromMilliseconds(500))
 .Subscribe(state => allowPublications = state);
 return Observable.Using(() => subscription, _ => deltaObservable.DoWhile(() => allowPublications));
 });

This is the simplest way I see of approaching your problem without exposing state. This allows you to completely encapsulate the state of your enabled/disabled delta publications. Replace startObservable, endObservable and deltaObservable as required. You may want to consider reducing the Throttle, but I figure it's pretty necessary since you are dealing with UI input to debounce your events.

You can expose combinedObservable and subscribe to that in place of your ManipulationDeltaEvent - the start/end capabilities will be handled automagically. You will need to add sequence repeating yourself.

• Veikedo: Actually events are raised by my WP7 controls. The goal of my code is to block swipe gesture in the pivot control.
• Veikedo: The goal of my code is to block a swipe gesture in the pivot control. The pivot (its name is Items in the code) is a child of page. To block swipe gesture I need to catch the ManipulationStarted event on Items and set Items.IsHitTestVisible = false to prevent handling swipe gesture. After Items, the event bubbled to the parent control (it is page). In the ManipulationDelta of page I complete "manipulation" (to prevent animation). Next I set Items.IsHitTestVisible = true to allow up\down (scrolling) gesture. So I'm not able to manually firing the events. Thank you for your reply!
• Dan: See my edit for a proposed solution.