如何逐个usnig rx和kotlin请求调用远程服务？

Question

我有应用程序必须向其发送数据的远程服务：改造中的定义2：

interface FooRemoteService {
    @POST("/foos")
    fun postFoos(@Body foos: List<FooPojo>): Observable<Response<List<String>>
 }

但是该调用一次的限制不超过X Foo。每个调用可以返回206个代码“部分成功”，其中包含不成功上传的Foo列表。还有413个“请求实体太大”。当然还有400和500。

并且应用程序需要发送未知数量的foo项目（由用户在运行时定义）。

为了避免服务应用程序的DDoS，需要一个接一个地发送此调用。

所以我在我的FooRepositoryImpl中实现了这样的功能：

这是一个想法。我对下面的解决方案不满意，我相信它可以做得更好，但我已经没有主意了。有什么建议吗？

override fun postFoos(foos: List<Foo>) Completable {
  val fooChunks = divideListInToChuncksUnderRequestLimit(foos)

  val unuploadedFoos = mutableListOf<UnuploadedFoo>()
  fooChunks.fold(unuploadedFoos) 
  { accu: MutableList<UnuploadedFoo>, chunk ->
     fooRemoteService
        .postFoos(chunk)
        .subscribeOn(Schedulers.io())
        .flatMapCompletable {
             if (it.isSuccessful) {
                         Completable.complete()
                    } else {
                        Timber.e("$it")
                       accu.add(it.body())

                    }
                }.blockingAwait()
        responses

  }
return Completable.complete()
}

在最后，应用程序应该显示所有不成功的fos的列表，或者如果有可用的。所以我需要从未上传的fos的功能列表中传递。

Answer 1

如果您可以稍微修改一下postFoos的返回类型，那么类似的操作可能会奏效：

override fun postFoos(foos: List<Foo>): Observable<List<UnuploadedFoo>> {
    val chunks = foos.chunked(CHUNK_SIZE)
    val posters = chunks.map { chunk ->
        fooRemoteService.postFoos(chunk)
                .map { response ->
                    response.unUploaded.takeIf { !response.isSuccessful } ?: emptyList()
                }
                .filter { it.isNotEmpty() }
                .toObservable()
    }

    return Observable.concatDelayError(posters)
}

我想象你的服务有这样的东西：

data class Response(val isSuccessful: Boolean, val unUploaded: List<UnoploadedFoo>)

fun postFoos(foos: List<Foo>): Single<Response>

这里的诀窍在于：

(...) 等待订阅传递给它的每个附加可观察对象，直到前一个可观察对象完成。