RxJava：当错误被吞下时，上游永远不会完成答案

【问题标题】：RxJava: upstream never completes when error is swallowedRxJava：当错误被吞下时，上游永远不会完成
【发布时间】：2018-10-02 23:51:06
【问题描述】：

我正在使用 RxJava 遍历文件列表，进行网络调用以上传每个文件，然后将成功上传的文件收集到列表中，并在成功时将这些文件保存在订阅者中。

此代码有效，但发生错误时除外。行为应该是它记录错误并继续，它会这样做，除非发生错误，订阅者的 onSuccess lambda 永远不会被调用。

观察者是否期望发射与原始迭代中相同数量的元素？迭代完所有项目后，如何跳过错误并使其完成？除了Single.never() 之外，还有其他东西可以实现不将错误转发到下游吗？

queryFiles()?.let { files ->
    Observable.fromIterable(files)
            .flatMapSingle { file ->
                uploadFile(file)
                        .onErrorResumeNext { error ->
                            log(error)
                            Single.never() // if this is returned onSuccess is never called
                        }
                        .map { response ->
                            file.id = response.id
                            file
                        }
            }
            .toList()
            .subscribe( { uploadedFiles ->
                persist(uploadedFiles) // if error occurs above, this is never called
            }, { error ->
                log(error)
            })
}

【问题讨论】：

标签： android kotlin rx-java rx-java2

【解决方案1】：

您的问题是Single 只能产生两个值，一个成功的结果或一个失败的结果。将失败转为“忽略”状态可以先将其转换为Maybe，然后使用基本相同的代码来处理失败和成功。

返回值为Maybe.empty() 的Maybe.onErrorResumeNext 将产生0 或1 个结果，而Maybe.map 仅在有值时执行，如您所描述的那样准确处理问题。

改编代码：

        .flatMapMaybe { file ->
            uploadFile(file).toMaybe()
                    .onErrorResumeNext { error: Throwable ->
                        log(error)
                        Maybe.empty()
                    }
                    .map { response ->
                        file.id = response.id
                        file
                    }
        }

【讨论】：

谢谢，效果很好。由于方法重载不明确，需要对上述代码进行的唯一更改是显式指定 error: Throwable 的类型。

【解决方案2】：

这是我过去使用zip 方法处理它的方式。

  // create an observable list that you can process for you file uploads
  val responses: Response = listOf<Response>()

  queryFiles()?.let { file ->

    val observable = Observable.create(ObservableOnSubscribe<Response> { emitter ->
      // you can modify this section to your data types
      try {
        // with your uploadFile method you might be able to just add them
        // all the responses list
        emitter.onNext(uploadFile(file))
        emitter.onComplete()
      } catch (e: Exception) {
        emitter.onError(e)
      }
    }).subscribeOn(Schedulers.io()).observeOn(AndroidSchedulers.mainThread())
    responses.add(observable)    
  }

  // i setup a simple booleanArray to handle success/failure but you can add
  // all the files that fail to a list and use that later    
  val isSuccessful = booleanArrayOf(true)
  Observable.zip<Response, Boolean>(responses, Function<Array<Any>, Boolean> { responses ->
    var isSuccessful: Boolean? = java.lang.Boolean.TRUE
    // handle success or failure
    isSuccessful
  }).subscribe(Consumer<Boolean> { aBoolean -> isSuccessful[0] = aBoolean!! }, Consumer<Throwable> { throwable ->
    isSuccessful[0] = false
  }, Action {
    // handle your OnComplete here
    // I would check the isSuccessful[0] and handle the success or failure        
  })

这会将您所有的上传内容创建到一个可以使用zip 方法处理和合并的 Observable 列表中。这将在它们完成后将它们全部合并到任何数组中，以便您可以循环它们 - 您来自 uploadFile() 方法的结果。此示例从返回的响应中检查成功或失败。我删除了注释// handle success or failure 所在的大部分逻辑。在函数方法中，您可以跟踪失败或成功的文件上传。

【讨论】：