Rx noobie here: any idea on how to achieve this?:
I have a list of items from the UI and would like to POST them to the server. I need the list of POSTed items back from the server (to use the server-issued IDs, timestamps etc).
Pardon my long examples but here's the code I'm struggle to make work:
/**
* 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>
}
UPDATE
I have tried the approach suggested by @AlexeySoshin below but still have some points of confusion:
I have tried the second, shorter approach but the retrofit call doesn't go thru for some reason :(, i.e the network service endpoints aren't hit). It's slightly modified because I couldn't get rx-kotlin wrapper to work, but I imagine it's roughly equivalent to this:
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)
}
}
My network service works successfully for a single Item with this code:
// 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))
}
How to make Flowable<Maybe<Item>>
calls go thru successfully? (or Flowable<List<Item>>
, which would be closer to what I need eventually)
Finally, how to obtain the actual output list from a
Flowable<Maybe<Item>>
? It would help to see how the .subscribe()
block might look like to 'extract' that final list. Here is how my current subscribe code looks like:
...
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 *//})
}
...