Rx：等待第一项一段时间

Question

我想将我遗留的基于事件的方法转换为基于可观察的方法，但我对 Rx 很陌生，所以我现在被卡住了。

我有一个事件源，现在可以观察到。在某个时间点，我必须启动一个方法，该方法要么通过返回行中的下一个元素结束，如果超时则返回 null。

基于事件的方法如下所示：

public async Task<ReaderEvent> WaitForReaderAsync(int PlaceId, TimeSpan waitFor)
{
    ReaderEvent result = null;
    using (var cts = CancellationTokenSource.CreateLinkedTokenSource(new [] { topLevelToken }))
    {
        cts.CancelAfter(waitFor);

        EventHandler<ReaderEvent> localHandler = (o, e) =>
        {
            if (e.PlaceId == PlaceId)
            {
                result = e;
                cts.Cancel();
            }
        };

        ReaderEventHandler += localHandler;
        try
        {
            await Task.Delay(waitFor, cts.Token).ConfigureAwait(false);
        }
        catch (OperationCanceledException) { }
        catch (Exception ex)
        {
            //...
        }

        ReaderEventHandler -= localHandler;
    }

    return result;
}

如您所见，这个想法是延迟被我正在等待的事件的到来取消，或者令牌源在该特定时间后被配置取消。挺干净的。

现在，Rx 版本：

public async Task<ReaderEvent> WaitForReaderAsync(int PlaceId, TimeSpan waitFor)
{
    ReaderEvent result = null;

    var observable = _OnReaderEvent.FirstAsync(r => r.PlaceId == PlaceId);

    using (var cts = CancellationTokenSource.CreateLinkedTokenSource(new [] { topLevelToken }))
    {
        cts.CancelAfter(waitFor);
        using (observable.Subscribe(x => {
            result = x;
            cts.Cancel();
        {
            try
            {
                await Task.Delay(waitFor, cts.Token).ConfigureAwait(false);
            }
            catch (OperationCanceledException) { }
        }
    }
    return result;
}

不太干净……更糟…… 我也尝试过超时扩展。但由于这是一次订阅，我仍然需要等待一段时间才能处理订阅。唯一的区别是 OnError 会取消本地令牌，而不是 CancelAfter 的内置机制。

有没有更好/更简洁（更依赖Rx）的方法来做到这一点？

谢谢！

Answer 1

你可以试试：

var values = await _OnReaderEvent
  .Where(r => r.PlaceId == placeId)
  .Buffer(waitFor, 1)
  .FirstAsync(); // get list of matching elements during waitFor time

return values.FirstOrDefault(); // return first element or null if the list is empty

Answer 2

为什么不使用简单的 Rx 版本代码：

public async Task<ReaderEvent> WaitForReaderAsync(int PlaceId, TimeSpan waitFor)
{
    return await
        _OnReaderEvent
            .Where(r => r.PlaceId == PlaceId)
            .Buffer(waitFor, 1)
            .Select(xs => xs.FirstOrDefault())
            .FirstOrDefaultAsync()
            .ToTask();
}

Answer 3

这个问题可以用很多不同的方法解决。这是一个，使用 Amb、Return、Delay 和 FirstAsync 运算符：

public Task<ReaderEvent> WaitForReaderAsync(int PlaceId, TimeSpan waitFor)
{
    return _OnReaderEvent
        .Where(r => r.PlaceId == PlaceId)
        .Amb(Observable.Return(default(ReaderEvent)).Delay(waitFor))
        .FirstAsync()
        .ToTask();
}

在 _OnReaderEvent observable 完成或等待期间完成的异常情况下，生成的 Task 将转换为故障状态，但 InvalidOperationException 序列除外不包含任何元素。

另一种实现，功能上等同于前一个，使用 Timeout 运算符：

public Task<ReaderEvent> WaitForReaderAsync(int PlaceId, TimeSpan waitFor)
{
    return _OnReaderEvent
        .Where(r => r.PlaceId == PlaceId)
        .FirstAsync()
        .Timeout(waitFor, Observable.Return(default(ReaderEvent)))
        .ToTask();
}

Rx：等待第一项一段时间

Rx: Wait for first item for a period of time

c#

task-parallel-library

system.reactive