How to cancel a composed RxJS observable

Question

Folks, I have an app using RxJS to handle mouse events. I am composing these events into more complex observable 'gestures'. One such gesture is "shake".

The series of events I am trying to compose are:

mousedown  
mousemove left  
mousemove right  
mousemove left  
mousemove right  
mouseup

What I am finding is that

mousedown  
mouseup
mousemove left  
mousemove right  
mousemove left  
mousemove right  

is also triggering the same result.

I have made a fiddle demonstrating the issue on codepen.

My question in general is: How do you express in RxJS that a obeservable, such as mouseup, should cancel and restart the composition of an observable?

As another example of the issue (from the same fiddle), the following events

mousedown  
mouseup
mousedown  
mouseup
mousedown  
mousemove left  
mousemove right  
mousemove left  
mousemove right  

Result in 3 shake events. I would like it to result in one.

Answer 1

Use takeUntil

var down = $div.onAsObservable('mousedown');
var up = $div.onAsObservable('mouseup');
var move = $div.onAsObservable('mousemove');

var shake = down.flatMapLatest(function (downev) {
            var current = downev;
            return move
                .map(function (ev) {
                    var delta = ev.pageX - current.pageX;
                    current = ev;
                    return delta < 0 ? 'left' : 'right';
                 })
                .takeUntil(up)
                .toArray();
        })
        .subscribe(function (moves) {
            // see if moves is an array of left,right,...