RxJava：在列表中创建项目并返回新列表

Question

Rx noobie here：关于如何实现这一点的任何想法？：

我有一个来自 UI 的项目列表，并希望将它们发布到服务器。我需要从服务器返回的已发布项目列表（以使用服务器发布的 ID、时间戳等）。

请原谅我的冗长示例，但这是我努力工作的代码：

/**
 * create each item (POST) in the list and return the list
 * of newly created ones
 */
fun postItems(items: List<Item>): Maybe<List<Item>> {
    // loop through all items, making a network call for each and return the
    // list of POSTed items

    // attempt 1
    // Failed type inference. Expected type mismatch:
    // Expected: Maybe<List<Item>>
    // Found: List<Maybe<Item>>
    return items.map {
        postItem(it)
    }

    // attempt 2: 'extract' each item in resulting observable and
    // shove it back to original list
    return Maybe.just(items.map {
        postItem(it!!)
                // 'extract' item from observable
                .flatMap {
                    // error: Type mismatch.
                    // Required ((Item) -> MaybeSource<out (???...???)>!)!
                    // Found (Item) -> Item
                    it
                }
    })

    // attempt 3: convert a Maybe to Flowable to collect all items from post
    // and emit them as a single list
    // Type mismatch. Required:(((Mutable)List<Item!>) -> SingleSource<out (???...???)>!)!
    // Found: ((Mutable)List<Item!>) -> (Mutable)List<Item!>
    return items.forEach {
        postItem(it!!).toFlowable().toList().flatMap { it }
    }

    // attempt 4: modify attempt 3 with concatMap:
    // Type mismatch. Required:((List<Item!>) -> MaybeSource<out (???...???)>!)!
    // Found: (List<Item!>) -> List<Maybe<Item>>
    return Maybe.just(items)
            // wait for each observable to finish all the work 
            // until the next one is processed
            .concatMap({
                it.map { addItem(it!!) }
            }).toFlowable()
            .toList().toMaybe()

    // attempt 6: blocking call on each API request.
    // results in android.os.NetworkOnMainThreadException
    return Maybe.just(places.map {
        addPlace(it!!).blockingGet()
    })
}

fun postItem(item: Item): Maybe<Item> {
    return networkService.post(item) // async call to API, returns Maybe<Item>
}

更新

我尝试了下面@AlexeySoshin 建议的方法，但仍有一些困惑：

我尝试了第二种更短的方法，但是由于某种原因，改造调用没有通过:(即没有命中网络服务端点)。它稍作修改，因为我无法让 rx-kotlin 包装器工作，但我想它大致相当于这个：

fun addItems(items: List<Item?>): Flowable<Maybe<Item>> {
    return Flowable.fromIterable(items).map {
        // tried items.toFlowable().map but got 'Unresolved reference toFlowable
        // even after adding rx-kotlin to project via gradle
        return@map postItem(it)
    }
}

我的网络服务使用此代码成功地为单个项目工作：

// works as expected
fun postOneItem(item: Item): Maybe<Item> {
    return postItem(item)
}

// also works
fun postOneItemFlowable(item: Item): Flowable<Item> {
    return postItem(item).toFlowable()
}


// this variant didn't work
fun postOneItemFlowable(item: Item): Flowable<Maybe<Item>> {
    return Flowable.just(postItem(item))
}

如何让Flowable<Maybe<Item>> 通话成功？ （或Flowable<List<Item>>，这将更接近我最终需要的）

最后，如何从一个 Flowable<Maybe<Item>>?这将有助于了解.subscribe() 块可能看起来像“提取”最终列表。这是我当前的订阅代码的样子：

...
private fun createItem(item: Item) {
    disposables.add(
            addItemUseCase.postOneItem(item)
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe({
                        Timber.d("one item posted: $it")
                    }, { /* error handler *//})
}
...

Answer 1

你可以这样做：

fun postItems(items: List<Item>): Maybe<List<Item>> {
    return items.toFlowable().map {
        postItem(it).blockingGet()
    }.toList().toMaybe()
}

但我不确定这就是你的真正意思，因为你的 Maybe 实际上不是 Maybe。

像这样更改签名会更有意义：

fun postItems(items: List<Item>): Flowable<Maybe<Item>> {
    return items.toFlowable().map {
        postItem(it)
    }
}