java - RxJava : Create items in list and return new list

Question

这里的 Rx 菜鸟:知道如何实现这一点吗？:

我有一个来自 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>> ，这将更接近我最终需要的)

最后，如何从a中获取实际的输出列表 Flowable<Maybe<Item>> ？了解 .subscribe() 会有所帮助。 block 可能看起来像是“提取”最终列表。这是我当前的订阅代码的样子:

...
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)
    }
}