使用 RxJava 的多个 POST 请求

Question

我有一个书籍清单。

open class Book(
    @PrimaryKey
    var id: String? = null,
    var title: String? = null,
    var author: String? = null
): RealmObject()

在 for 循环时，我过滤了一些书并创建了一个带有过滤后的 Observable。我将每个 observable 添加到一个数组中。

val listInserts = ArrayList<Observable<Book>>()
for loop(..) {
  if (localBook condition) {
    val postObservable = networkApiAdapter.insert(localBook)
    listInserts.add(postObservable)
  }
}

我合并（或连接）可观察对象，希望有一些连续的 POST 请求。

        Observable.concat(listInserts)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe(
                { s ->
                    println(s)
                },
                { err ->
                    println(err)
                },
                { println("onComplete") }
            )

在我的 MongoDb+Flask 服务器上总是只收到一个 POST 请求，我也收到此错误：

com.google.gson.JsonSyntaxException: java.lang.IllegalStateException: Expected BEGIN_OBJECT but was NUMBER at line 1 column 4 path $

networkApiAdapter 函数，改造：

class NetworkAPIAdapter private constructor() {
    fun insert(dto: Book): Observable<Book> {
        println(dto.toString())
        return bookService.insert(dto.title!!, dto.author!!)
    }

    interface BooksService {
        @FormUrlEncoded
        @POST(URL_ORDERS_ALL)
        fun insert(
            @Field("title") title: String,
            @Field("author") author: String
        ): Observable<Book>
    }
}

欢迎任何帮助。我不知道如何做多个请求。我用 zip、repeatUntil、flatMap 尝试了许多解决方案，但都没有奏效。

解决此问题后，我必须删除所有本地书籍并执行 GET 请求。所有这些都使用 RxJava 工作。

Answer 1

您的BooksService.insert 被定义为POST 请求，该请求期望服务器以Book JSON 字符串（即Observable<Book>）的形式作出响应。我认为这不是您的服务器在 POST 请求通过后返回的内容。检查您的POST 请求返回什么，我认为它返回一些Int（可能是响应状态）而不是Book，这可以解释您的错误：

com.google.gson.JsonSyntaxException: java.lang.IllegalStateException: Expected BEGIN_OBJECT but was NUMBER at line 1 column 4 path $

如果是这种情况，您的串联 Observable 将在抛出错误后立即终止，即在您的第一个 POST 请求之后，因此将不会执行剩余的请求。

如果我的假设是正确的，请将您的服务更改为：

interface BooksService {
        @FormUrlEncoded
        @POST(URL_ORDERS_ALL)
        fun insert(
            @Field("title") title: String,
            @Field("author") author: String
        ): Observable<Int>
    }

另外，如果你想让你的其他请求即使前一个抛出错误也能执行，这样做：

val postObservable = networkApiAdapter.insert(localBook).onErrorReturn(_ -> SOME_INT_FLAG)

更新

要在发送完所有POST 请求后执行您需要的逻辑，请尝试：

    Observable.concat(listInserts)
        // complete this observable after all POST requests have been sent
        // not that it does not necesarily mean that all responses were 200, you should implement yourself what happens to items that were not successfully `POST`ed
        .take(listInserts.size)
        .doOnComplete { 
            // called when this observable completes, i.e. when all POST requests have been sent
            executeFinalActions()
        }
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
            { s->
                println(s)
            },
            { err ->
                println(err)
            },
            { println("onComplete") }
        )

并实现executeFinalActions()函数：

private fun executeFinalActions() {
    realm.executeTransaction(Realm::deleteAll)
    networkApiAdapter.fetchAll()
    .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe(
            { books ->
                // todo: store books
            },
            { err ->
                println(err)
            }
        )
}

PS：你可能想看看Observable.take() 做了什么：http://reactivex.io/documentation/operators/take.html