我试图把我的头包裹在RX周围但我的大脑爆炸了:)

我想要做的是通过RX以异步方式调用WCF方法.
这里没什么特别的,但是当WCF方法抛出一个错误时,我想重新创建一个通道并再给它一次(最多3次).

到目前为止我所拥有的是：

var _sc = new Service.Service1Client();
var _observableFunc = Observable.FromAsyncPattern<int, string>(_sc.BeginGetData, _sc.EndGetData);
var _observable = _observableFunc(666);

var _defered = Observable.Defer(() => _observable);

// Here something should be done, but don't know what...

using (_retryable.Subscribe(x => Console.WriteLine("Async ==> '{0}'", x),
                            ex => Console.WriteLine("Oops ==> {0}", ex.Message)))
{
    Console.ReadLine();
}

我玩过Catch< TSource,TException>,它允许我捕获异常并继续使用相同的observable,因此给了我想要的东西.
唯一的问题是它会永远运行,这意味着如果我继续抛出异常,事情永远不会停止！

解决方法:

试着这样做：

var retryable = Observable.Defer(() => _observableFunc(666).Retry(3));

重试扩展方法“将源可观察序列重复指定的次数或直到它成功终止”.

另外,不要这样做：

var _observable = _observableFunc(666);
var _defered = Observable.Defer(() => _observable);

在您将它开除后,没有必要推迟观察.

你应该这样做：

var _defered = Observable.Defer(() => _observableFunc(666));

然后你离我在顶部的建议解决方案只有一步之遥.