如何使用 Rx.Net 实现状态轮询？

Question

我已经玩了 2 天了，现在尝试使用 .Net 的响应式扩展来拥抱响应式编程领域。

我构建了一个状态轮询用例，假设一个虚拟 Web API 和一个轮询状态对象的反应式客户端。

我尝试了以下代码：

        // Creates an observable that ticks each 1 second
        var ticksObservable = Observable.Interval(TimeSpan.FromMilliseconds(1000));



        // Creates a new observable transforming each tick to a string status requested from the api
        var coldStatusPollerObservable = ticksObservable.Select(tick =>
        {
            Console.WriteLine("Sending Request");
            var tsk = client.GetStatus(1); // Http get request to a web api resource (id == 1 just for demo)
            tsk.Wait();
            return tsk.Result;

        }
        );

       // Subscribe and print results on console
       coldStatusPollerObservable.Subscribe(
        status => Console.WriteLine(status), ex => Console.WriteLine(ex.Message)
        );

一切都很好，我得到了预期的输出：

{"status":"waiting"}
{"status":"running"}
{"status":"running"}
{"status":"running"}
{"status":"ok"}

然后我添加了另一个约束，即从 Web API 返回的随机错误请求。 发生的问题是我无法正确处理异常。 异常发生在 tsk.wait() 中，我期望它只会触发我传递给订阅方法 ( ex => Console 的 onError 操作.WriteLine(ex.Message) )

Q1：在这种情况下处理异常的正确方法是什么？ Q2：使用 Rx.NET 是否有更简洁的轮询实现？

PS：我使用的是 Rx.NET 3.1.1

Answer 1

如果可以的话，您基本上希望避免使用任何 .Wait() 阻塞调用。 Rx 带有一个专门用于处理任务的运算符 - Observable.FromAsync

所以你的基本查询现在变成了：

var coldStatusPollerObservable =
    from tick in ticksObservable
    from status in Observable.FromAsync(() => client.GetStatus(1))
    select status;

如果您想要控制台消息，请执行以下操作：

var coldStatusPollerObservable =
    from tick in ticksObservable
    from status in Observable.FromAsync(() =>
    {
        Console.WriteLine("Sending Request");
        return client.GetStatus(1);
    })
    select status;

请记住，尽可能坚持使用内置运算符。

---

您可以像这样处理异常：

void Main()
{
    var coldStatusPollerObservable =
        from tick in Observable.Interval(TimeSpan.FromMilliseconds(1000))
        from status in
            Observable
                .FromAsync(() => client.GetStatus(1))
                .Catch<string, Exception>(ex => Observable.Return("Error"))
        select status;

    coldStatusPollerObservable.Subscribe(x => Console.WriteLine(x), ex => Console.WriteLine(ex.Message));
}

public static class client
{
    private static int _counter = 0;
    public static Task<string> GetStatus(int id)
    {
        if (_counter++ == 5)
            throw new Exception();
        return Task.Run(() => _counter.ToString());
    }
}

这给出了：

1 2 3 4 5 错误 7 8 9 10 11 12 13 14 15 ...