.net - 如何转换 IObservable使用 CancellationToken 支持在 RX 中列出

Question

我有一个方法返回一个IObservable<long>我从一个方法调用的async方法。我想将其转换为 normal List<long>，但如果我收到CancellationToken信号则取消该操作。

我想做这样的事情：

List<long> result = await Foo().ToList(myCancellationToken);

实现此目的的正确（和最简单）方法是什么？的ToList()扩展方法IObservable<T>返回一个IObservable<List<T>>并且不带CancellationToken参数。

Answer 1

var list = await Foo().ToList().ToTask(cancellationToken);

This has the advantage of cancelling immediately if the token is cancelled (the other Answer will not cancel until the next time the Foo observable produces a value).

Answer 2

使用 TakeWhile 终止列表。

CancellationToken MyToken = ...
var list = await Foo().TakeWhile(v=>!MyToken.IsCancellationRequested).ToList();

如果您担心只有在提供下一个项目时才会取消订阅，您可以使用此扩展方法。

public static IObservable<T> 
TakeWhile<T>
    ( this IObservable<T> This
    , CancellationToken t 
    )
{
    var cts = CancellationTokenSource.CreateLinkedTokenSource(t);
    return Observable.Create<T>(( IObserver<T> observer ) => 
       { 
           This.Subscribe(observer, cts.Token);
           return Disposable.Create(() => cts.Cancel());
    });
}

和写

CancellationToken MyToken = ...
var list = await Foo().TakeWhile(MyToken.IsCancellationRequested).ToList();

使用带有取消标记的 TakeWhile 比仅返回最后一个元素的 ToTask 更具可组合性。