替换代表状态的 Rx.NET Subject 的方法

如何解决替换代表状态的 Rx.NET Subject 的方法

我目前正在使用以下方法修复错误，该方法在 stateChecker 时轮询 null 条件，直到它变为 true（或 false 由于超时）：>

private static void WaitWithSubject(
   Func<bool> stateChecker,TimeSpan timeout,TimeSpan stepTime,string errorMessage,ILifetimeInfo lifetimeInfo)
{
   (bool? IsOk,string Message) state = (IsOk: null,Message: string.Empty);
   var waitCancellation = (int)stepTime.TotalMilliseconds;
   using (var stateSubject = new Subject<(bool? IsOk,string Message)>())
   {
      using (Observable.Timer(timeout).Subscribe(it => stateSubject.OnNext((IsOk: false,Message: errorMessage))))
      using (Observable.Timer(TimeSpan.Zero,stepTime).
         Subscribe(it =>
         {
            if (stateChecker())
               stateSubject.OnNext((IsOk: true,Message: string.Empty));
         }))
      {
         using (stateSubject.Subscribe(it => state = it))
         {
            while (state.IsOk == null)
               lifetimeInfo.Canceler.ThrowIfCancellationRequested(waitCancellation);
            if (state.IsOk != true)
               throw new TimeoutException(state.Message);
            stateSubject.OnCompleted();
         }
      }
   }
}

此方法偶尔会在执行方法 ObjectdisposedException 的代码中的以下点生成 OnNext :

if ( stateChecker() )
    stateSubject.OnNext( ( IsOk: true,Message: string.Empty ) );

在这种情况下，有没有办法完全避免使用 Subject 以支持诸如 Observable.Interval 或 Observable.Create 之类的东西？

解决方法

在我看来，这就是你想要做的：

private static void WaitWithSubject(Func<bool> stateChecker,TimeSpan timeout,TimeSpan stepTime,string errorMessage,ILifetimeInfo lifetimeInfo) =>
    Observable
        .Amb(
            Observable
                .Timer(timeout)
                .SelectMany(_ => Observable.Throw<Unit>(new TimeoutException(errorMessage))),Observable
                .Timer(TimeSpan.Zero,stepTime)
                .Where(_ => stateChecker())
                .Select(_ => Unit.Default))
        .Take(1)
        .Wait();

这里的关键是 Amb 运算符，它启动两个序列，并且只返回第一个序列的值以产生值或错误。 Take(1) 确保可观察对象在产生值后立即完成。

如果您有 Wait()，您可以在 CancellationToken 之前添加以下行以取消：

.TakeUntil(Observable.Create<Unit>(o => ct.Register(() => o.OnNext(Unit.Default))))

---

在与 Theodor 反复讨论之后，我想出了这个我认为可能是我能想到的最干净的版本：

private static void WaitWithSubject(Func<bool> stateChecker,ILifetimeInfo lifetimeInfo)
{
    var good =
        Observable
            .Timer(TimeSpan.Zero,stepTime)
            .Where(_ => stateChecker())
            .Take(1);

    var fail =
        Observable
            .Timer(timeout)
            .SelectMany(_ => Observable.Throw<long>(new TimeoutException(errorMessage)));
    
    good.Merge(fail).RunAsync(lifetimeInfo.Canceler).Wait();
}

,

这是一种与 WaitWithSubject 具有类似行为的方法，但不使用 Subject<T>。它使用 Merge 运算符，以便将两个计时器生成的序列合并为一个序列。它还具有取消支持。

public static void WaitUntilTrueState(
    Func<bool> stateChecker,TimeSpan checkInterval,string timeoutMessage,CancellationToken cancellationToken = default)
{
    Observable
        .Timer(TimeSpan.Zero,checkInterval)
        .Merge(Observable.Timer(timeout).IgnoreElements()
            .Concat(Observable.Throw<long>(new TimeoutException(timeoutMessage))))
        .TakeUntil(_ => stateChecker())
        .RunAsync(cancellationToken)
        .Wait();
}