Mouse input with Rx - Making a sequence of observables?

Question

I'm an Rx newbie trying to figure out how to process mouse gestures with Rx. I found this solution somewhere:

var mouseMove = Observable.FromEventPattern<MouseEventArgs>(this, "MouseMove");
var lMouseDown = Observable.FromEventPattern<MouseEventArgs>(this, "MouseDown")
    .Where(e => e.EventArgs.Button == MouseButtons.Left);
var lMouseUp = Observable.FromEventPattern<MouseEventArgs>(this, "MouseUp")
    .Where(e => e.EventArgs.Button == MouseButtons.Left);
var dragSequence =
    from down in lMouseDown
    from move in mouseMove.StartWith(down).TakeUntil(lMouseUp)
    select move;
dragSequence.ObserveOn(this).Subscribe(e => Trace.WriteLine(e.EventArgs.Location));

But multiple independent mouse gestures are all part of the same stream. So I can't use a handler for onCompleted; the sequence is never completed. I'd like to break the stream into a separate sequence for each drag, how do I do that?

Answer 1

Here's my solution:

var mouseMove = Observable.FromEventPattern<MouseEventArgs>(this, "MouseMove");
var lMouseDown = Observable.FromEventPattern<MouseEventArgs>(this, "MouseDown")
    .Where(e => e.EventArgs.Button == MouseButtons.Left);
var lMouseUp = Observable.FromEventPattern<MouseEventArgs>(this, "MouseUp")
    .Where(e => e.EventArgs.Button == MouseButtons.Left);

lMouseDown.SelectMany(start =>
{
    // a new drag event has started, prepare to receive input
    var dragSeq = new List<Point>();
    Action<EventPattern<MouseEventArgs>, bool> onNext = (e, mouseUp) => {
        // This code runs for each mouse move while mouse is down.
        // In my case I want to constantly re-analyze the shape being
        // drawn, so I make a list of points and send it to a method.
        dragSeq.Add(e.EventArgs.Location);
        AnalyzeGesture(dragSeq, mouseUp);
    };

    return mouseMove
        .StartWith(start)
        .TakeUntil(lMouseUp.Do(e => onNext(e, true)))
        .Do(e => onNext(e, false));
})
.Subscribe();

How this works is, each time a mouse-down arrives, the start=>{...} lambda runs. This lambda returns an observable that uses Do() to process each input. Notice that the lambda itself creates an event stream without subscribing to it, and I am discarding the results of the inner and outer observable, because Do() has already processed the input.

The lambda does not subscribe to the query because the outer Subscribe() has the effect of subscribing to both an individual mouse drag and the whole sequence of them (thanks to SelectMany).

In case the mouse-up point differs from last mouse move, I used Do() to capture it. However, it appears that the mouse-up point is always equal to the previous point. So here's a slightly simpler version that ignores the mouse-up point:

lMouseDown.SelectMany(start =>
{
    var dragSeq = new List<Point>();
    return mouseMove
        .StartWith(start)
        .TakeUntil(lMouseUp)
        .Do(e => {
            dragSeq.Add(e.EventArgs.Location);
            AnalyzeGesture(dragSeq, false);
        }, () => AnalyzeGesture(dragSeq, true));
})
.Subscribe();