代码之家 › 专栏 › 技术社区 › Jeff Lockhart

RxJava:当错误被吞没时,上游永远不会完成

rx-java2 rx-java kotlin android

Jeff Lockhart · 技术社区 · 7 年前

观察者是否期望发射与原始iterable中相同数量的元素?我如何跳过错误并在所有项目都被迭代之后完成它?有别的事吗 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)
            })
}

2 回复 | 直到 7 年前

Kiskae 7 年前

你的问题是 Single 只能产生两个值,成功或失败。将故障转换为“忽略”状态可以通过首先将其转换为 Maybe

Maybe.onErrorResumeNext 返回值为 Maybe.empty() 将导致0或1个结果,而 Maybe.map 只有当它有一个值时才执行,并按照您所描述的那样准确地处理问题。

改编代码:

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

Ray Hunter 7 年前

下面是我过去使用 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        
  })

这是将您的所有上传创建到一个可观察的列表中,可以用 拉链 方法。当它们被合并到一个任意数组中时,这将把它们全部合并起来,这样您就可以在它们上面循环—您从uploadFile()方法得到的结果。本例检查返回的响应是否成功。我删除了评论中的大部分逻辑 // handle success or failure 是。在函数方法中,您可以跟踪失败或成功的文件上载。