How to interrupt existing IObservable sequence?

Question

I have

I have an observable sequence made from touch events:

 var touchEvents = Observable.Merge(
     Observable.FromEventPattern<RoutedEventArgs>(touchRect, nameof(touchRect.PointerEntered)),
     Observable.FromEventPattern<RoutedEventArgs>(touchRect, nameof(touchRect.PointerExited)),
     Observable.FromEventPattern<RoutedEventArgs>(touchRect, nameof(touchRect.PointerReleased)));

I observe until a specific condition is met:

async void Start()
{
    _scanSlapObservable = touchEvents.Throttle(TimeSpan.FromSeconds(1))
                    .Where(e => countOfPointers == expectedFingers.Length)
                    .Select(e => new Response())
                    .FirstOrDefaultAsync();

    Response response = await _scanSlapObservable;
    if (response != null)
    { 
       //do something
    }
}

What I need

When I call Start() twice, I want to interrupt the existing _scanSlapObservable, so it either returns null or throws an exception.

How do I interrupt the existing IObservable sequence?

Possible duplicate of [How to cancel an observable sequence](http://stackoverflow.com/questions/6759833/how-to-cancel-an-observable-sequence) — Benjamin Podszun, Dec 14 '16 at 14:33
This seems to be a duplicate (marked as such). While you don't use the ```Generate``` methods to .. generate the observable (hence cannot pass a CancellationToken), you might be able to insert a ```TakeUntil()``` and have another Observable for your interruption event? — Benjamin Podszun, Dec 14 '16 at 14:35
Its not duplicate since the proposed answer won't work for me. — Liero, Dec 15 '16 at 09:00
I'm affraid I will need better solution than TakeUntil, since I need to interrupt it immediatelly. TakeUntil will wait until next touch event occurs — Liero, Dec 15 '16 at 09:02
That is an incorrect assumption. The `TakeUntil` solution outlined below will fire immediately. — Shlomo, Dec 15 '16 at 15:58

score 0 · Answer 1 · answered Dec 14 '16 at 16:22

Here's a Linqpad friendly solution. Ugly, but it works:

async Task Main()
{
    var sc = new ScanClass();
    sc.Start();
    sc.touchEvents.OnNext(Unit.Default);
    sc.touchEvents.OnNext(Unit.Default);
    sc.Start();
    sc.touchEvents.OnNext(Unit.Default);
    sc.touchEvents.OnNext(Unit.Default);
    await Task.Delay(TimeSpan.FromSeconds(1));
}

// Define other methods and classes here
class ScanClass
{
    IObservable<Response> _scanSlapObservable;
    public Subject<Unit> _interrupter = new Subject<Unit>();
    public Subject<Unit> touchEvents = new Subject<Unit>();

    public async Task Start()
    {
        _interrupter.OnNext(Unit.Default);

        _scanSlapObservable = touchEvents
                        .Throttle(TimeSpan.FromSeconds(1))
//                      .Where(e => countOfPointers == expectedFingers.Length)
                        .Select(e => new Response())
                        .TakeUntil(_interrupter)
                        .FirstOrDefaultAsync();

        Response response = await _scanSlapObservable;
        if (response != null)
        {
            //do something
            Console.WriteLine("Do Something");
        }
        else
        {
            Console.WriteLine("Null Response");
        }
    }
}
class Response { }

I think the best way to go would be a Switch style though that wouldn't work with your Tasks. I think a large part of the ugliness is mixing the Task API with Rx. Here's what an Rx-only solution would look like:

class ScanClass
{
    public Subject<Unit> _starter = new Subject<Unit>();
    public Subject<Unit> touchEvents = new Subject<Unit>();

    public ScanClass()
    {
        _starter.Select(_ => touchEvents
            .Throttle(TimeSpan.FromSeconds(1))
//          .Where(e => countOfPointers == expectedFingers.Length)
            .Select(e => new Response())
        )
        .Switch()
        .Subscribe(r => Console.WriteLine("Do Something."));
    }

public async Task Start()
    {
        _starter.OnNext(Unit.Default);
    }
}

...and ideally you would replace Start() and _starter with some sort of Observable.FromEvent.

How to interrupt existing IObservable sequence?

I have

What I need

1 Answers1