How would I implement a sequence doing HTTP reques...

# coroutines

[ribesg]

How would I implement a sequence doing HTTP request using the Ktor client? I get an error about restricted suspending functions when using sequence { }

[gildor]

sequence doesn’t support any kind async events, this is syncronous only abstraction

[gildor]

you can use Channels instead

[ribesg]

Oh, ok

[gildor]

In future you also can use cold streams, when they will be released, for now Channels is the only abstraction for data streams compatible with coroutines

[ribesg]

I’m basically implementing pagination, I don’t know much about channels. Is there an example somewhere on how to use channels for that?

[gildor]

I would start first from reading official docs
https://github.com/Kotlin/kotlinx.coroutines/blob/master/docs/channels.md

[gildor]

briefly: instead of sequence { } use CoroutineScope.produce { }

[ribesg]

Yeah that’s what I get from those docs. So send would not be called until there’s a consumer, right?

[gildor]

Depends on channel type

[gildor]

there are different ones, but this is default one, rendezvous channel, and it works as you described

[Fredrik Larsen]

Sorry to highjack the thread, but you mentioned cold streams. Would they work similar to Rx observables?

[gildor]

@Fredrik Larsen Yes, the main idea is the same, but based on coroutines
More details:
https://github.com/Kotlin/kotlinx.coroutines/issues/254

[Fredrik Larsen]

Interesting. Thank you @gildor

[ribesg]

@gildor when using a producer, it does one more request than needed because it waits for a receive when I call send, but when I call send I already did the next request. Is there a way to start the next request when receive is called, and not before, to prevent downloading potentially useless data?

[bdawg.io]

I'm not entirely sure if onSend would get you that. @gildor do you happen to know if onSend is only a listener event or if it actually puts an item on the queue as well?

[bdawg.io]

Is there a way to start the next request when receive is called, and not before

fun CoroutineScope.loadPages() = produce<List<MyItem>> {
    var currentPage = 1
    while (true) {
        val page = loadPageNum(currentPage)
        if (page.isEmpty()) {
            break;
        }
        // await until item is received
        send(page)
        currentPage += 1
    }
}

[ribesg]

@bdawg.io this does download the next page before it’s needed. Also `onSend`’s prototype is scary

[bdawg.io]

@ribesg I'm not understanding why. Unless you're specifying a capacity on your producer, it should suspend until someone `receive`s it. That's why it's called a Rendezvous channel

[bdawg.io]

I've never used onSend haha so I wasn't sure if it'd actually be useful or not lol

[ribesg]

You’re calling loadPageNum before blocking on send

[ribesg]

So the page is loaded before we start waiting for someone to consume the page

[ribesg]

I don’t want to download it unless something wants it

[bdawg.io]

That sounds like a cold stream. An alternative is to return a Lazy<ReceiveChannel<YourType>> and then have your consumer unwrap the value before consuming

val pages by lazy { loadPages() }
pages.consumeEach { ... }

[ribesg]

That doesn’t change anything as we’re still loading one more page than needed.

[ribesg]

I totally need a cold stream here it seems, but there’s no such thing right now

[bdawg.io]

Why doesn't it change anything? Only one page is loaded at a time and it doesn't get loaded until first access of the Lazy (for above, not until pages.consumeEach is called)?

[ribesg]

@bdawg.io I’m not sure how to explain it better, but you always retrieve a page before calling send which is the blocking point. So you always have a unused page in memory, which may never be used. Your lazy trick only works on the first page.

[gildor]

@ribesg Yes, exactly, this is how hot streams (channels) work and this will be properly solved only by cold streams

[ribesg]

In the meantime, here’s my solution:

interface PageIterator<T> {
        suspend fun next(): List<T>
    }

object : PageIterator<MyType> {

            private var lastReturnedPage = -1
            private var done = false

            override suspend fun next(): List<MyType> {
                if (done) return emptyList()
                val res = doMyHttpRequest(++lastReturnedPage)
                done = lastReturnedPage == res.nbPages - 1
                return res.results
            }

        }

[gildor]

Sure, this is your own solution for this

[gildor]

Actually this problem is described in first and second comment in Cold Streams issue and one of main use cases

[ribesg]

Yes, I know that cold streams are the solution, but I needed an actual real solution for now, not a future one 🙂

[gildor]

It’s not so a future, planned for April, as I heard
But I agree, nothing wrong with own case-specific solution, especially if you don’t integrate with other data streams

[ribesg]

A lot of things I need will be out very soon, that’s unfortunate but I need to progress in my implementation

[gildor]

Again, I think it’s completely fine, some kind high level abstraction for streams required only if you have to manage multiple data streams and it may be much more tricky with ad-hoc solution

[ribesg]

Yeah, in my case I just needed a suspend next function doing an Http request, so I did just that. Hopefully I’ll be able to use existing things in the future.