1

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

1 Answer 1

0

You can do something like this:

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

But I'm not sure that's what you actually mean, since your Maybe is not a Maybe, actually.

Changing the signature like this will make much more sense:

fun postItems(items: List<Item>): Flowable<Maybe<Item>> {
    return items.toFlowable().map {
        postItem(it)
    }
}
5
  • This looks promising, but how come you have a List being converted to a Flowable with items.toFlowable()...? I thought such methods are only available on an Observable.
    – kip2
    Commented Oct 2, 2018 at 10:51
  • 1
    He's probably using rx-kotlin (you tagged your question with rx-kotlin yourself). toFlowable() is an extension function that calls Flowable.fromIterable()
    – arekolek
    Commented Oct 2, 2018 at 11:13
  • 1
    It's RxKotlin, and you can add it with compile 'io.reactivex.rxjava2:rxkotlin:2.3.0' Commented Oct 2, 2018 at 12:53
  • @AlexeySoshin I have updated the post to add more info about my experience trying your answer. I appreciate your patience :)
    – kip2
    Commented Oct 4, 2018 at 7:59
  • I cannot run your code, but having return@map seems wrong. Usually return@ is a quick return, like break sort of thing. I suggest that you try it with one call to server, and then improve it to work with map() of calls. Commented Oct 4, 2018 at 19:40

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.