非重放热可观察

原问题

我有一个场景，我有多个IObservable我想要组合的序列Merge然后听。但是，如果其中之一产生错误，我不希望它使其他流的所有内容崩溃，也不希望重新订阅序列（这是一个“永远持久”的序列）。

我通过附加一个来做到这一点Retry()合并之前的流，即：

IEnumerable<IObservable<int>> observables = GetObservables();

observables
    .Select(o => o.Retry())
    .Merge()
    .Subscribe(/* Do subscription stuff */);

然而，当我想测试这个时，问题就出现了。我想测试的是，如果其中之一IObservables in observables产生一个OnError，其他人应该仍然能够发送他们的值，并且他们应该得到处理

我以为我只用两个Subject<int>s代表两个IObservables in observables;一个发送一个OnError(new Exception())另一个，在那之后，发送OnNext(1)。然而，似乎Subject<int>将重播新订阅的所有先前值（这实际上是Retry()是），将测试变成无限循环。

我尝试通过创建手册来解决这个问题IObservable这会在第一个订阅上产生错误，然后产生一个空序列，但感觉很奇怪：

var i = 0;
var nErrors = 2;
var testErrorObservableWithOneErrorAndThenCompletion = Observable.Create<int>(o => {
    i++;
    if (i < nErrors) {
        return Observable.Throw<int>(new Exception()).Subscribe(o);
    } else {
        return Observable.Empty<int>().Subscribe(o);
    }
});

我正在使用吗Subject或思考Retry()以错误的方式？对此还有其他想法吗？你会如何解决这种情况？

Update

好的，这是我想要的大理石图think Retry() does.

o = message, X = error.
------o---o---X
               \
     Retry() -> \---o---o---X
                             \
                   Retry() -> \...

我的问题可能更多是因为我没有一个好的库存类来使用前测试，因为Subject想要重播我以前的所有错误。

Update 2

这是一个测试用例，显示了我的意思Subject重演它的价值观。如果我说它在 a 中这样做，我使用这个术语是否正确？cold方式？我知道Subject是创建热可观察值的一种方式，但这种行为对我来说仍然感觉“冷”。

var onNext = false;
var subject = new Subject<int>();

subject.Retry().Subscribe(x => onNext = true);
subject.OnError(new Exception());
subject.OnNext(1);

Assert.That(onNext, Is.True);

根据您更新的要求（您想要重试失败的可观察量，而不是只想忽略它们），我们可以提出一个可行的解决方案。

首先，了解冷可观察值（在每个订阅上重新创建）和热可观察值（无论订阅如何都存在）之间的区别很重要。你不能Retry()一个热门的可观察对象，因为它不知道如何重新创建底层事件。也就是说，如果一个热的可观察错误，它就会永远消失。

Subject创建一个热可观察对象，从某种意义上说，您可以调用OnNext没有订阅者，它将按预期运行。要将热可观察量转换为冷可观察量，您可以使用Observable.Defer，其中将包含该可观察对象的“订阅创建”逻辑。

话虽如此，这里是修改后的原始代码：

var success = new Subject<int>();
var error = new Subject<int>();

var observables = new List<IObservable<int>> { Observable.Defer(() => {success = new Subject<int>(); return success.AsObservable();}), 
                                               Observable.Defer(() => {error = new Subject<int>(); return error.AsObservable();}) };                                            

observables
.Select(o => o.Retry())
.Merge()
.Subscribe(Console.WriteLine, Console.WriteLine, () => Console.WriteLine("done"));

和测试（与之前类似）：

success.OnNext(1);
error.OnError(new Exception("test"));
success.OnNext(2);
error.OnNext(-1);
success.OnCompleted();
error.OnCompleted();

以及预期的输出：

1
2
-1
done

当然，您需要根据您底层可观察的内容来显着修改这个概念。使用受试者进行测试与实际使用它们不同。

我还想指出这个评论：

然而，似乎主题将重播所有以前的值 新的订阅（实际上是 Retry() ），将测试变成 无限循环。

事实并非如此——Subject不这样做。您的代码的其他一些方面导致了无限循环，基于以下事实：Retry重新创建订阅，并且订阅在某个时刻会产生错误。

原答案（用于完成）

问题是Retry()不做你想做的事。从这里：

http://msdn.microsoft.com/en-us/library/ff708141(v=vs.92).aspx http://msdn.microsoft.com/en-us/library/ff708141(v=vs.92).aspx

重复源可观察序列 retryCount 次或直到 它成功终止。

这意味着Retry将不断尝试重新连接到底层可观察对象，直到成功并且不会抛出错误。

我的理解是你实际上希望可观察到的异常是ignored，未重试。这将做你想做的事：

observables
.Select(o => o.Catch((Func<Exception,IObservable<int>>)(e => Observable.Empty<int>())))
.Merge()
.Subscribe(/* subscription code */);

这使用Catch捕获异常的可观察量，并在此时将其替换为空的可观察量。

这是使用主题的完整测试：

var success = new Subject<int>();
var error = new Subject<int>();

var observables = new List<IObservable<int>> { success.AsObservable(), error.AsObservable() };

observables
.Select(o => o.Catch((Func<Exception,IObservable<int>>)(e => Observable.Empty<int>())))
.Merge()
.Subscribe(Observer.Create<int>(Console.WriteLine, Console.WriteLine, () => Console.WriteLine("done")));

success.OnNext(1);
error.OnError(new Exception("test"));
success.OnNext(2);
success.OnCompleted();

正如预期的那样，这会产生：

1
2
done