所以我尝试使用onErrorReturn返回我想要的结果，但它之后会完成流，我如何捕获错误返回为 Next 并仍然继续流？
使用下面的代码，当出现错误时它不会达到retryWhen，如果我翻转它，它不会重新订阅retryWhen如果出现错误
fun process(): Observable<State> {
return publishSubject
.flatMap { intent ->
actAsRepo(intent) // Might return error
.map { State(data = it, error = null) }
}
.onErrorReturn { State(data = "", error = it) } // catch the error
.retryWhen { errorObs ->
errorObs.flatMap {
Observable.just(State.defaultState()) // continue subscribing
}
}
}
private fun actAsRepo(string: String): Observable<String> {
if (string.contains('A')) {
throw IllegalArgumentException("Contains A")
} else {
return Observable.just("Wrapped from repo: $string")
}
}
订户将是
viewModel.process().subscribe(this::render)
onError 是一个终端运算符。如果发生 onError，它将在操作员之间传递。您可以使用 onError-运算符来捕获 onError 并提供回退。
在你的例子中，onError 发生在 flatMap 的内部流中。onError 将传播到下游的 onErrorReturn opreator。如果你看一下实现，你会看到onErrorReturn lambda将被调用，结果将在onComplete之后被onNext推向下游
@Override
public void onError(Throwable t) {
T v;
try {
v = valueSupplier.apply(t);
} catch (Throwable e) {
Exceptions.throwIfFatal(e);
downstream.onError(new CompositeException(t, e));
return;
}
if (v == null) {
NullPointerException e = new NullPointerException("The supplied value is null");
e.initCause(t);
downstream.onError(e);
return;
}
downstream.onNext(v); // <--------
downstream.onComplete(); // <--------
}
您的解决方案的结果是什么？
您的流完成的原因如下： #retryWhen JavaDoc
如果运算符的上游是异步的，则立即发出 onNext 然后是 onComplete 的信号可能会导致序列立即完成。类似地，如果这个内部 {@code ObservableSource} 在上游处于活动状态时发出 {@code onError} 或 {@code onComplete} 信号，则序列将立即以相同的信号终止。
你应该做什么：
将 onErrorReturn 放在 flatMap 中的地图操作器后面。通过此排序，您的流将无法完成，当内部平面映射流打开错误时。
这是为什么呢？
当外部(源：publishSubject(和内部流(订阅(都完成时，flatMap 运算符完成。在这种情况下，外部流 (publishSubject( 发出 onNext，内部流将在通过 onNext 发送 { State(data = "， error = it( } 后完成。因此，流将保持打开状态。
interface ApiCall {
fun call(s: String): Observable<String>
}
class ApiCallImpl : ApiCall {
override fun call(s: String): Observable<String> {
// important: warp call into observable, that the exception is caught and emitted as onError downstream
return Observable.fromCallable {
if (s.contains('A')) {
throw IllegalArgumentException("Contains A")
} else {
s
}
}
}
}
data class State(val data: String, val err: Throwable? = null)
apiCallImpl.call 将返回一个延迟的可观察量，这将在订阅时抛出错误，而不是在可观察程序集时引发错误。
// no need for retryWhen here, except you want to catch onComplete from the publishSubject, but once the publishSubject completes no re-subscription will help you, because the publish-subject is terminated and onNext invocations will not be accepted anymore (see implementation).
fun process(): Observable<State> {
return publishSubject
.flatMap { intent ->
apiCallImpl.call(intent) // Might return error
.map { State(data = it, err = null) }
.onErrorReturn { State("", err = it) }
}
}
测试
lateinit var publishSubject: PublishSubject<String>
lateinit var apiCallImpl: ApiCallImpl
@Before
fun init() {
publishSubject = PublishSubject.create()
apiCallImpl = ApiCallImpl()
}
@Test
fun myTest() {
val test = process().test()
publishSubject.onNext("test")
publishSubject.onNext("A")
publishSubject.onNext("test2")
test.assertNotComplete()
.assertNoErrors()
.assertValueCount(3)
.assertValueAt(0) {
assertThat(it).isEqualTo(State("test", null))
true
}
.assertValueAt(1) {
assertThat(it.data).isEmpty()
assertThat(it.err).isExactlyInstanceOf(IllegalArgumentException::class.java)
true
}
.assertValueAt(2) {
assertThat(it).isEqualTo(State("test2", null))
true
}
}
另类
此替代方法的行为与第一个解决方案略有不同。flatMap-Operator 采用布尔值 (delayError(，这将导致吞噬 onError 消息，直到源完成。源完成后，将发出错误。
您可以使用 delayError true，当异常没有用并且必须在出现时记录
时过程
fun process(): Observable<State> {
return publishSubject
.flatMap({ intent ->
apiCallImpl.call(intent)
.map { State(data = it, err = null) }
}, true)
}
测试
仅发出两个值。错误不会转换为回退值。
@Test
fun myTest() {
val test = process().test()
publishSubject.onNext("test")
publishSubject.onNext("A")
publishSubject.onNext("test2")
test.assertNotComplete()
.assertNoErrors()
.assertValueAt(0) {
assertThat(it).isEqualTo(State("test", null))
true
}
.assertValueAt(1) {
assertThat(it).isEqualTo(State("test2", null))
true
}
.assertValueCount(2)
}
注意：我认为在这种情况下你想使用switchMap，而不是flatMap。