取消RX.Net Observer的正在进行的OnNext方法

Question

正如我原来的问题所述（请参阅Correlate interdependent Event Streams with RX.Net），我有一个RX.net事件流，只有在某个其他事件没有被触发时才会调用观察者的OnNext方法（基本上是'处理更改 - *只要系统已连接，事件就会中断，断开连接时暂停，并在系统重新连接后重新开始处理Change- *事件。

但是，虽然这项新事件顺利进行，但我将如何取消/取消信号到正在进行中 .OnNext（）调用？

Answer 1

由于您的观察者已经接受CancellationToken，我们可以修改您的Rx流以提供一个事件数据。我们将使用Rx CancellationDisposable，只要流取消订阅，我们将处置该Rx CancellationDisposable。

// Converts into a stream that supplies a `CancellationToken` that will be cancelled when the stream is unsubscribed 
public static IObservable<Tuple<CancellationToken, T>> CancelOnUnsubscribe<T>(this IObservable<T> source) 
{ 
    return Observable.Using(
     () => new CancellationDisposable(), 
     cts => source.Select(item => Tuple.Create(cts.Token, item))); 
} 

与其他问题的解决方案放在一起这样的：

DataSourceLoaded 
    .SelectMany(_ => DataSourceFieldChanged 
     .Throttle(x) 
     .CancelOnUnsubscribe() 
     .TakeUntil(DataSourceLoaded)) 
    .Subscribe(c => handler(c.Item1, c.Item2)); 

当TakeUntil条款被触发，它将从CancelOnUnsubscribe观察到的取消，将在CancellationDisposable和事业转处置令牌被取消。您的观察员可以观看此令牌并在发生此情况时停止其工作。

Answer 2

有一个异步超载SelectMany，虽然承认如果Do存在类似的过载，它会更符合语义。

var subscription = 
    (from _ in DataSourceLoaded 
    from __ in DataSourceFieldChanged 
    .Throttle(x) 
    .SelectMany(DataSourceFieldChangedAsync) 
    .TakeUntil(DataSourceUnloaded) 
    select Unit.Default); 
    .Subscribe(); // Subscribing for side effects only. 

... 

async Task<Unit> DataSourceFieldChangedAsync(Field value, CancellationToken cancel); 

这是不错的，因为它关系到取消的订阅为好。

调用要么

subscription.Dispose() 

或

DataSourceUnloaded.OnNext(x); 

将导致CancellationToken被取消。