c# - 如何中断现有的 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)));

我观察直到满足特定条件:

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。我认为很大一部分丑陋是将 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);
    }
}

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