如何中断现有的iObservable序列

Question

我有

我有一个由触摸事件制成的可观察序列：

 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观察到满足特定条件为止：

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
    }
}

我需要的

当我两次调用Start()时，我想中断现有的_scanSlapObservable，因此它返回null或抛出异常。

如何中断现有的iObservable序列？

Answer 1

这是一个linqpad友好解决方案。丑陋，但有效：

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 { }

我认为最好的方法是Switch样式，尽管它与您的Task s不起作用。我认为丑陋的很大一部分是将Task API与RX混合。这是仅RX的解决方案的样子：

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);
    }
}

...理想情况下，您将用某种Observable.FromEvent替换Start()和_starter。