And I only want to call

request

if my buffer is not full, i.e. I can actually take more elements in without just storing them in memory.

So I can't just suspend in

onNext

. I could block the thread, but that defeats the point.

Current code (simplified):

class CallbackImpl<V>(private val scope: ProducerScope<V>) : CallbackIface<V> {
  override fun onNext(value: V) {
    check(scope.channel.offer(value)) // channel is unlimited, to not loose messages
    request(1) // <= I only want to do this if the channel actually has room
  }
  override fun onCompleted() {
    scope.channel.close()
  }
}
fun adaptApi(): Flow<V> {
  return callbackFlow<V> {
    api.doStuff(CallbackImpl(this))
    awaitClosed()
  }.buffer(Channel.UNLIMITED)
}

Now that I write this out in a simplified form I also noticed that I then also need to know if the channel becomes exhausted, so I need to request more elements again...

So what I want (pseudocode):

channel.runWhenBufferLessThan(2) { api.request(1) }

This would be "easy" if the backpressure was enforced. However

request

is only a hint, and the API may call

onNext

at any time, in which case I need to buffer the elements.

Otherwise this would work for me: https://github.com/vert-x3/vertx-lang-kotlin/blob/2243a9c1c2323f27dd31d0d18e4979cd7b945c51/vertx-lang-kotlin-coroutines/src/main/java/io/vertx/kotlin/coroutines/ReceiveChannelHandler.kt#L123-L159

If I do the equivalent, I am just launching potentially infinite coroutines if my consumer is slow and the producer doesn't adhere to the backpressure

Current idea:

callbackFlow

with

buffer(RENDEZVOUS)

. Inside create another channel, which has unlimited buffering. Launch a coroutine which polls the buffered channel. If it finds an element, it sends it on to the

callbackFlow

channel (likewise with close/exception). If

poll

returns null that means we need to request more elements, so then we call

request

. Then in my callback

onNext

I just do

offer

on the buffered channel (which will always succeed without blocking, because it's an unlimited channel). Does this sound even remotely reasonable? It feels very weird having two channels...

Time for bed. Maybe my brain will come up with something better while I sleep.

Have a

flow { }

which creates a channel and spawns a coroutine, then simply read from the channel and emit directly in

flow

.

This is my code now:

return flow {
        val bufferedChannel = Channel<Result>(Channel.UNLIMITED)

        val callback = CallbackAdapter(bufferedChannel)
        apiCall(observer)

        // request initial element
        observer.request(1)

        // wait for first element, suspending
        while (bufferedChannel.transferOneSuspend(this@flow)) {
            // keep transferring while the buffer has elements
            while (bufferedChannel.transferOneIfBuffered(this@flow)) {}

            // buffer ran dry, request more elements
            callback.request(requestSize)
            // and now go back to suspending transfer until more arrive
        }
    }

It uses two helper functions:

/**
 * Transfer one element from this `ReceiveChannel` into the given `FlowCollector` if it is possible to do so
 * without suspending to receive.
 *
 * If this channel has failed, this method will throw the causing exception.
 * If no element is currently buffered in this channel or the channel is
 * [closed for receive][ReceiveChannel.isClosedForReceive], this method will return false.
 * Otherwise this method returns `true`.
 */
suspend fun <V : Any> ReceiveChannel<V>.transferOneIfBuffered(target: FlowCollector<V>): Boolean {
    val element = poll()
    return if (element == null) {
        false
    } else {
        target.emit(element)
        true
    }
}

/**
 * Transfer one element from this `ReceiveChannel` into the given `FlowCollector`.
 * If this channel has failed, this method will throw the causing exception.
 * If this channel is closed, this method will return `false`.
 * If an element was successfully transferred, this method will return `true`.
 */
@ExperimentalCoroutinesApi
suspend fun <V : Any> ReceiveChannel<V>.transferOneSuspend(target: FlowCollector<V>): Boolean {
    val element = receiveOrNull()
    return if (element == null) {
        // we are done
        false
    } else {
        target.emit(element)
        true
    }
}

I am quite happy with it.