替换表示状态的Rx.NET主题的方法

Question

我目前正在修复以下方法中的错误，该方法在null时轮询stateChecker条件，直到它变成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();
}
}
}
}

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

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

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

Answer 1

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

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)确保一旦产生一个值，可观察的就结束。

如果您有CancellationToken:，您可以在Wait()之前插入以下行以取消

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

---

在与西奥多进行了一番反复之后，我提出了这个版本，我认为它可能是我能想到的最干净的版本：

private static void WaitWithSubject(Func<bool> stateChecker, TimeSpan timeout, TimeSpan stepTime, string errorMessage, 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();
}

Answer 2

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

public static void WaitUntilTrueState(
Func<bool> stateChecker,
TimeSpan checkInterval,
TimeSpan timeout,
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();
}