With kotlinx-coroutines-io-jvm 0.1.0-beta1 I used ...

# coroutines

[rocketraman]

With kotlinx-coroutines-io-jvm 0.1.0-beta1 I used to have code that did ByteReadChannel.lookAheadSuspend in which I was able to call consumeEachRemaining, like this:

fun CoroutineScope.flowableOf(byteReadChannel: ByteReadChannel): Flowable<ByteBuffer> = rxFlowable {
  byteReadChannel.lookAheadSuspend {
    consumeEachRemaining {
      send(it.copy())
      true
    }
  }
}

In kotlinx-coroutines-io 0.1.0, this same code compiles but does not work -- it does not read the entire channel. Neither does the corresponding non-suspending lookAhead method work.

[orangy]

@cy

[rocketraman]

Trying it now with reverting to 0.1.0-beta-1, but staying on ktor 1.0.0-rc

[rocketraman]

Nope, doesn't look like those two are compatible

[cy]

Let me look at it

[cy]

It doesn't do awaitAtLeast

[cy]

You can always write a loop instead

[rocketraman]

What doesn't do awaitAtLeast? The 0.1.0 library code?

[cy]

consumeEachRemaining

[cy]

there are no suspension points at all in it's loop

[rocketraman]

But its the same code as 0.1.0-beta-1 isn't it? AFAICT the library implementation is the same.

[cy]

Then it worked accidentally

[cy]

See how it is used in ktor: https://github.com/ktorio/ktor/blob/1.0.0/ktor-network/src/io/ktor/network/sockets/CIOWriter.kt#L62

[cy]

In ktor it is used to write directly to NIO socket in CIO engine

[rocketraman]

Ok, is there a simpler way to consume all of a ByteReadChannel? I wouldn't expect this to be very difficult.

[cy]

Most likely consumeEachRemaining need to be suspend function. For now you need a loop

[rocketraman]

Ok, let me try to replicate the logic in ktor

[rocketraman]

Now I got too many bytes, lol

[cy]

This is relatively dangerous API, but in your case I think it is acceptable to use it. For regular users I wouldn't recommend to use it at all

[cy]

If you have too many bytes, perhaps you forgot consumed(N)

[cy]

Or have it specified wrong

[rocketraman]

Yup, I forgot consumed

[rocketraman]

Hmm: java.lang.IllegalStateException: Unable to consume 4088 bytes: not enough available bytes

[rocketraman]

lookAheadSuspend {
    // consumeEachRemaining does not work for now, use the loop logic from ktor
    while (true) {
      val buffer = request(0, 1)
      if (buffer == null) {
        if (!awaitAtLeast(1)) break
        continue
      }

      while (buffer.hasRemaining()) {
        val count = buffer.remaining()
        send(buffer.copy())
        consumed(count)
      }
    }
  }

[rocketraman]

Exception thrown from consumed(count)

[cy]

You actually don't need the second loop

[cy]

What exception?

[rocketraman]

java.lang.IllegalStateException: Unable to consume 4088 bytes: not enough available bytes

[cy]

Does copy modify buffer's position?

[rocketraman]

Yes it does seem to, its copy from Ktor

[cy]

Well, firs of all, it's ktor internal function, you shouldn't use it anymore

[cy]

The second thing is that it keeps position untouched so you should have infinite loop here

[cy]

It's not infinite just because consumed fails as you are trying to consume the same bytes twice

[cy]

In fact you don't need the second loop at all

[rocketraman]

Fair enough... ok, this is what I have now, which seems to work in a single test case:

while (true) {
      val buffer = request(0, 1)
      if (buffer == null) {
        if (!awaitAtLeast(1)) break
        continue
      } else {
        val count = buffer.remaining()
        send(buffer.duplicate())
        consumed(count)
      }
    }

[rocketraman]

duplicate is just the bog-standard JDK ByteBuffer method

[cy]

This will lead to serious damage

[cy]

duplicate doesn't copy buffer content but creates a view

[rocketraman]

Oh, right, it shares content

[cy]

You need your own copy function instead

[rocketraman]

Ok, this is what I have now:

[rocketraman]

lookAheadSuspend {
    // consumeEachRemaining does not work for now, use the loop logic from ktor
    while (true) {
      val buffer = request(0, 1)
      if (buffer == null) {
        if (!awaitAtLeast(1)) break
        continue
      } else {
        val count = buffer.remaining()
        send(buffer.copy())
        consumed(count)
      }
    }
  }

and

fun ByteBuffer.copy(): ByteBuffer {
  val readBuffer = asReadOnlyBuffer()
  return ByteBuffer.allocate(readBuffer.remaining()).apply {
    put(readBuffer)
    flip()
  }
}

[rocketraman]

Seems to work but any comments?

[cy]

continue is redundant since you have else. However it should work

[rocketraman]

Hang on, above you said ktor internal copy leaves buffer position untouched. My version does the same. I think if we want to actually consumer the input channel (which in this case, I think it should be), then my copy method should be:

fun ByteBuffer.copy(): ByteBuffer {
  return ByteBuffer.allocate(remaining()).apply {
    put(this@copy)
    flip()
  }
}

... and yes, this works too.

[cy]

there is no difference since you don't have a loop anymore

[rocketraman]

Ah, the buffer.hasRemaining -- right

[rocketraman]

So there is no problem with moving the position on these ktor-provided buffers right?

[rocketraman]

And also, do you want me to create an issue for consumeEachRemaining?

[cy]

Yes, please

[cy]

And yes, the problem was in the loop condition so now there is no problem with moving buffer's position

[cy]

The only requirement is to not share a buffer returned from request

[cy]

it is even not required to consume all bytes from the buffer but you can't consume more bytes than you actually have

[rocketraman]

FYI: https://github.com/Kotlin/kotlinx-io/issues/22