多用户ReactiveX重试

Question

快问，因为我觉得我一定是漏掉了什么。我在这里使用rxjs，因为这是我前面的内容，我相信这是一个一般的reactiveX问题。

假设我有一组这样的可观察到的：

network_request = some_thing // An observable that produces the result of a network call
event_stream = network_request.flatMapLatest(function(v) {
    return connectToThing(v) // This is another observable that needs v
}) // This uses the result of the network call to form a long-term event-based connection

所以，这还行。不过有问题。有时连接失败。

所以，如果我做了event_stream.retry()，效果很好。当它失败时，它会重做网络调用，并获得一个新的v来建立一个新的连接。

问题

如果我希望将两件事情从我的network_request中链接起来，会发生什么呢？也许我希望每次网络调用完成时UI都能做些什么，比如在UI中显示一些关于v的内容？

我能做到：

shared = network_request.share() // Other implementations call this refCount
event_stream = shared.flatMapLatest(...) // same as above
ui_stream = shared.flatMapLatest(...) // Other transformation on network response

如果我不执行share，那么它会发出两个请求，这不是我想要的，但是使用share时，当event_stream稍后出错时，它不会重试网络请求，因为refcount仍在1(由于ui_stream)，因此它立即返回because。

我想要的

这显然是我编造的一个小例子来解释我的困惑。我想要的是，每次event_stream (长期连接)的结果出现错误时，都会发生以下情况：

1. 再次发出网络请求。
2. 该请求的新响应用于构建新连接，event_stream继续处理新事件，就像什么都没有发生一样。
3. 在ui_stream中也会发出同样的响应，以导致进一步的处理。

这并不是一件复杂的事情，所以当涉及到拆分/扇出RX事物时，我一定是误解了一些基本的东西。

我想我可以做些什么，但我想避免。

我想出口这些可观察到的东西，所以我不能再建一次，然后说“嘿，这是新事物”。我希望event_stream和所有下游处理都不知道已经断开了连接。ui_stream也一样。它只是得到了一个新的价值。

我可能会用一个Subject作为代计数器，每次我想要重新启动所有东西时，我都会打开它，并在此基础上将network_request放入一个flatMap中，这样我就可以打破share.但这感觉像一个非常棘手的解决方案，所以我觉得必须有一个更好的方法。

我从根本上误解了什么？

Answer 1

当我更多地思考这个问题的时候，我得到了和电离一样的认识，那就是retry只是断开和重新连接，而上游不知道这是因为一个错误。

当我想到我想要什么的时候，我意识到我真的想要一个像链子一样的东西，也想要一个旁观者，所以我现在有这样的想法：

network_request = some_thing
network_shadow = new Rx.Subject()

event_stream = network_request.do(network_shadow).flatMapLatest(...)
ui_stream = network_shadow.whatever

在这个属性中，event_stream或下游的重试将导致整个事件重新启动，而ui_stream是它自己的事情。那里的任何错误都不会做任何事情，因为network_shadow实际上不是event_stream的订阅者，但是只要主事件链运行，它就会剥离值。

我觉得这并不理想，但它比我所担心的要做的要好，doOnError中有一个doOnError，它会很恶心。

我现在要处理这件事，看看它在哪里咬我.

Answer 2

你需要用Publish让你的冷能被观察到。阅读HotAndColdObservables.html#HotAndCold获得一个很好的解释。