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

Question

正如我最初的问题中所述（请参阅Correlate interdependent Event Streams with RX.Net），我有一个 RX.net 事件流，只要不触发某个其他事件（基本上是“将更改-* 事件处理为只要系统已连接，则在断开连接时暂停，并在系统重新连接后重新开始处理 Change-* 事件。

但是，虽然这适用于新事件，但我将如何取消/向正在进行的 .OnNext() 调用发出取消信号？

Answer 1

由于您的观察者已经被写入接受CancellationToken，我们可以修改您的 Rx 流以提供一个连同事件数据。我们将使用 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 observable，这反过来又会处理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 被取消。