.net 如何在发布前将取消令牌插入ReactiveX流(IObservable)？

wz1wpwve 于 2023-05-02 发布在 .NET

关注(0)|答案(1)|浏览(133)

如何在调用Publish之前将取消令牌插入现有的IObservable管道（即即，在它变成IConnectableObservable之前）？
在订阅之前，这必须是冷可观察管道的一部分（否则，我可以将CancellationToken令牌传递给IObservable的Subscribe，RunAsync，ToTask等）。
有没有推荐的模式？
我可以考虑使用TakeUntil来实现这一点，正如Theodor Zoulias here所建议的那样。例如：

using System.Reactive.Linq;
using System.Reactive.Threading.Tasks;

async Task Test(CancellationToken token)
{
    var publishedSequence = Observable
        .Interval(TimeSpan.FromMilliseconds(100))
        .Do(n => Console.WriteLine($"Emitting: {n}"))
        .Skip(3)
        .TakeUntil(
            Observable.Create<long>(
                observer => token.Register(
                    (_, token) => observer.OnError(new OperationCanceledException(token)),
                    null)))
        .Finally(() => Console.WriteLine($"Finally"))
        .Publish();

    using var subscription = publishedSequence.Subscribe(
            onNext: n => Console.WriteLine($"OnNext: {n}"),
            onError: e => Console.WriteLine($"OnError: {e}"),
            onCompleted: () => Console.WriteLine("OnCompleted"));

    using var connection = publishedSequence.Connect();
    await publishedSequence.ToTask();
}

var cts = new CancellationTokenSource(1000);
await Test(cts.Token);

输出：

Emitting: 0
Emitting: 1
Emitting: 2
Emitting: 3
OnNext: 3
Emitting: 4
OnNext: 4
Emitting: 5
OnNext: 5
Emitting: 6
OnNext: 6
Emitting: 7
OnNext: 7
Emitting: 8
OnNext: 8
OnError: System.OperationCanceledException: The operation was canceled.
Finally

我还有一个自定义运算符WithCancellation的原型，它基本上是一个直通IObservable，也在监听抵消信号。我宁愿坚持一个标准的方法。

更新，我想我在TakeUntil的工作方式中发现了一个竞争条件（可能是一个bug或只是一个我无法解释的行为），fiddle。我不能用我自己开发的WithCancellation实现（在fiddle中注解掉了）来重现它。

更新了，如果我使用.TakeUntil(Task.Delay(Timeout.Infinite, token).ToObservable())，我也不能复制它。

.net

来源：https://stackoverflow.com/questions/72471152/how-to-plug-a-cancellation-token-into-a-reactivex-stream-iobservable-before-pu

1条答案

按热度按时间

e4yzc0pl1#

实际上，没有用于此的库方法。我将从CancellationToken创建一个observable，然后使用TakeUntil操作符。

public static IObservable<Unit> ToObservable(this CancellationToken cancellationToken) {
  // if the token can't be cancelled, use Never which will not complete
  if (!cancellationToken.CanBeCanceled) {
    return Observable.Never<Unit>();
  }

  // if the token is already cancelled, use Return which publishes
  // Unit and completes immediately.
  // This may save you from ObjectDisposedException later
  if (cancellationToken.IsCancellationRequested) {
    return Observable.Return<Unit>();
  }

  // use Create so that each .Subscribe is handled independently
  return Observable.Create<Unit>(observer => {

    // Observable.Create does not handle errors on its own
    try {
      // return the registration because Dispose will unregister it
      return cancellationToken.Register(() => {
        // When the token is cancelled, publish and complete
        observer.OnNext(Unit.Default);
        observer.OnCompleted();
      });
    }
    catch (ObjectDisposedException e) {
      // todo: consider handling this as if it were cancellation
      observer.OnError(e);
    }
  });
}

public static IObservable<T> TakeUntil<T>(this IObservable<T> source, CancellationToken cancellationToken)
  => source.TakeUntil(cancellationToken.ToObservable());

赞(0）回复(0）举报 2023-05-02

我来回答

.net 如何在发布前将取消令牌插入ReactiveX流(IObservable)？

1条答案

相关问题

热门标签

最新问答