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

Question

Rx noobie这里:对如何实现这一点有什么想法吗？：

我有一个来自UI的项目列表，并希望将它们发布到服务器。我需要从服务器返回的POSTed项目列表(使用服务器发布的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下面建议的方法，但仍然有些困惑：

我尝试了第二种更短的方法，但由于某些原因，更新调用没有被执行:(即网络服务端点没有被击中)。由于无法让roughly包装器工作，所以对它进行了稍微的修改，但我认为它大致相当于以下内容：

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