将生成的事件序列创建为冷序列

FWIW -在询问了meta

的建议后，我将放弃这个问题的先前版本，以相同的方式支持不同的版本。

我有一个包含配置数据的web服务。我希望定期调用它Tok，以便刷新使用它的应用程序中的配置数据。如果服务出错(超时，down等)，我想保留以前调用的数据，并在不同的时间间隔Tnotok后再次调用服务。最后，我希望行为是可测试的。

由于管理时间序列和可测试性似乎是响应式扩展的强项，我开始使用一个可观察对象，它将由生成的序列提供。下面是我创建序列的方法:

Observable.Generate<DataProviderResult, DataProviderResult>(
    // we start with some empty data
    new DataProviderResult() { 
            Failures = 0
            , Informations = new List<Information>()},
    // never stop
    (r) => true,
    // there is no iteration
    (r) => r,
    // we get the next value from a call to the webservice
    (r) => FetchNextResults(r),
    // we select time for next msg depending on the current failures
    (r) => r.Failures > 0 ? tnotok : tok,
    // we pass a TestScheduler
    scheduler)
.Suscribe(r => HandleResults(r));

我现在有两个问题:

---

看起来我正在创建一个热门的可观察对象。即使尝试使用Publish/Connect，我所描述的动作也错过了第一个事件。如何将其创建为冷可观察对象?

myObservable = myObservable.Publish();
myObservable.Suscribe(r => HandleResults(r));
myObservable.Connect() // doesn't call onNext for first element in sequence

---

当我订阅时，订阅和生成的顺序似乎不一致，因为对于任何帧，订阅方法在FetchNextResults方法之前触发。这正常吗?我希望序列调用frame f的方法，而不是f+1。下面是我用来获取和订阅的代码:

private DataProviderResult FetchNextResults(DataProviderResult previousResult)
{
    Console.WriteLine(string.Format("Fetching at {0:hh:mm:ss:fff}", scheduler.Now));
    try
    {
        return new DataProviderResult() { Informations = dataProvider.GetInformation().ToList(), Failures = 0};
    }
    catch (Exception)
    {}
    previousResult.Failures++;
    return previousResult;
}
private void HandleResults(DataProviderResult result)
{
    Console.WriteLine(string.Format("Managing at {0:hh:mm:ss:fff}", scheduler.Now));
    dataResult = result;
}

这是我所看到的，促使我提出这些问题:

Starting at 12:00:00:000
Fetching at 12:00:00:000 < no managing the result that has been fetched here
Managing at 12:00:01:000 < managing before fetching for frame f
Fetching at 12:00:01:000
Managing at 12:00:02:000
Fetching at 12:00:02:000

---

编辑:这里有一个简单的复制-粘贴程序来说明这个问题。

/*using System;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using Microsoft.Reactive.Testing;*/
private static int fetchData(int i, IScheduler scheduler)
{
  writeTime("fetching " + (i+1).ToString(), scheduler);
  return i+1;
}
private static void manageData(int i, IScheduler scheduler)
{
  writeTime("managing " + i.ToString(), scheduler);
}
private static void writeTime(string msg, IScheduler scheduler)
{
  Console.WriteLine(string.Format("{0:mm:ss:fff} {1}", scheduler.Now, msg));
}
private static void Main(string[] args)
{
    var scheduler = new TestScheduler();
    writeTime("start", scheduler);
    var datas = Observable.Generate<int, int>(fetchData(0, scheduler),
                                                (d) => true,
                                                (d) => fetchData(d, scheduler),
                                                 (d) => d,
                                                 (d) => TimeSpan.FromMilliseconds(1000),
                                                 scheduler)
                                                 .Subscribe(i => manageData(i, scheduler));
    scheduler.AdvanceBy(TimeSpan.FromMilliseconds(3000).Ticks);
}

输出如下内容:

00:00:000 start
00:00:000 fetching 1
00:01:000 managing 1
00:01:000 fetching 2
00:02:000 managing 2
00:02:000 fetching 3

我不明白为什么第一个元素的管理没有在抓取后立即拾取。在有效提取数据的序列和将数据传递给观察者之间有一秒钟的时间。是我遗漏了什么，还是这是预期行为?如果是这样，是否有办法让观察者立即对新值做出反应?

private DataProviderResult FetchNextResult()
{
    // let exceptions throw
    return dataProvider.GetInformation().ToList();
}
private IObservable<DataProviderResult> CreateObservable(IScheduler scheduler)
{
    // an observable that produces a single result then completes
    var fetch = Observable.Defer(
        () => Observable.Return(FetchNextResult));
    // concatenate this observable with one that will pause
    // for "tok" time before completing.
    // This observable will send the result
    // then pause before completing.
    var fetchThenPause = fetch.Concat(Observable
        .Empty<DataProviderResult>()
        .Delay(tok, scheduler));
    // Now, if fetchThenPause fails, we want to consume/ignore the exception
    // and then pause for tnotok time before completing with no results
    var fetchPauseOnErrors = fetchThenPause.Catch(Observable
        .Empty<DataProviderResult>()
        .Delay(tnotok, scheduler));
    // Now, whenever our observable completes (after its pause), start it again.
    var fetchLoop = fetchPauseOnErrors.Repeat();
    // Now use Publish(initialValue) so that we remember the most recent value
    var fetchLoopWithMemory = fetchLoop.Publish(null);
    // YMMV from here on.  Lets use RefCount() to start the
    // connection the first time someone subscribes
    var fetchLoopAuto = fetchLoopWithMemory.RefCount();
    // And lets filter out that first null that will arrive before
    // we ever get the first result from the data provider
    return fetchLoopAuto.Where(t => t != null);
}
public MyClass()
{
    Information = CreateObservable();
}
public IObservable<DataProviderResult> Information { get; private set; }

Observable.Generate<DataProviderResult, DataProviderResult>(
    // we start with some empty data
    new DataProviderResult() { 
            Failures = 0
            , Informations = new List<Information>()},
    // never stop
    (r) => true,
    // we get the next value(iterate) by making a call to the webservice
    (r) => FetchNextResults(r),
    // there is no projection
    (r) => r,
    // we select time for next msg depending on the current failures
    (r) => r.Failures > 0 ? tnotok : tok,
    // we pass a TestScheduler
    scheduler)
.Suscribe(r => HandleResults(r));