IObservable以无限循环产生结果

Question

这是我到目前为止开发的代码:

var observable = Observable.Create<string>(async observer =>
{
    var wc = new WebClient { UseDefaultCredentials = true };
    observer.OnNext(await wc.DownloadStringTaskAsync("http://ya.ru"));
});

observable.Subscribe(
    res => Debug.WriteLine("got result: {0}", res), 
    exc => Debug.WriteLine("exception: {0}", exc.Message)
); 

这正确地获取网站的数据并触发我的回调一次.我想要的是有一个无限循环,其行为如下:等待结果 - >调用OnNext- >等待n秒 - >重复一个操作.

创建一个Observable.Interval和SelectMany我的Observable意志不会做,因为这将在一段固定的时间内查询网站.我希望下一次调用仅在上一次成功或失败后触发.实现这一目标的最优雅,最简洁的方法是什么？

Answer 1

在不改变代码的情况下,您可以在收益后有效地延迟延迟,然后无限期地重复整个事情.

var observable = Observable.Create<string>(async observer =>
{
    var wc = new WebClient { UseDefaultCredentials = true };
    observer.OnNext(await wc.DownloadStringTaskAsync("http://ya.ru"));
});

observable
    .Concat(Observable.Empty<string>().Delay(TimeSpan.FromSeconds(1)))
    .Repeat()
    .Subscribe(
      res => Debug.WriteLine("got result: {0}", res), 
      exc => Debug.WriteLine("exception: {0}", exc.Message)
    ); 

但是,有更好的方法可以做到这一点,因为在之前的实例中,您每秒都在创建一个新的WebClient.相反,你可以做这样的事......

using System.Reactive.Threading.Tasks;

var observable = Observable.Using(() => new WebClient(), (client) =>
    client.DownloadStringTaskAsync("http://ya.ru")
        .ToObservable()
        .Concat(Observable.Empty<string>().Delay(TimeSpan.FromSeconds(1)))
        .Repeat()
        );

如果您想重复错误,可以添加Retry...

var observable = Observable.Using(() => new WebClient(), (client) =>
    client.DownloadStringTaskAsync("http://ya.ru")
        .ToObservable()
        .Retry(3)
        .Concat(Observable.Empty<string>().Delay(TimeSpan.FromSeconds(1)))
        .Repeat()
        );