With kotlinx coroutines io jvm 0 1 0 beta1 I used to have co

Question

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 lt ByteBuffer gt = 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 · Answer

< cy>

rocketraman · Answer

Trying it now with reverting to 0 1 0 beta 1 but staying on ktor 1 0 0 rc

rocketraman · Answer

Nope doesn t look like those two are compatible

cy · Answer

Let me look at it

cy · Answer

It doesn t do `awaitAtLeast`

cy · Answer

You can always write a loop instead

rocketraman · Answer

What doesn t do `awaitAtLeast` The 0 1 0 library code

cy · Answer

`consumeEachRemaining`

cy · Answer

there are no suspension points at all in it s loop

rocketraman · Answer

But its the same code as 0 1 0 beta 1 isn t it AFAICT the library implementation is the same

cy · Answer

Then it worked accidentally

cy · Answer

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 · Answer

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

rocketraman · Answer

Ok is there a simpler way to consume all of a `ByteReadChannel` I wouldn t expect this to be very difficult

cy · Answer

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

rocketraman · Answer

Ok let me try to replicate the logic in ktor

rocketraman · Answer

Now I got too many bytes lol

cy · Answer

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 · Answer

If you have too many bytes perhaps you forgot `consumed N `

cy · Answer

Or have it specified wrong

rocketraman · Answer

Yup I forgot `consumed`

rocketraman · Answer

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

rocketraman · Answer

``` 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 · Answer

Exception thrown from `consumed count `

cy · Answer

You actually don t need the second loop

cy · Answer

What exception

rocketraman · Answer

`java lang IllegalStateException Unable to consume 4088 bytes not enough available bytes`

cy · Answer

Does `copy` modify buffer s position

rocketraman · Answer

Yes it does seem to its `copy` from Ktor

cy · Answer

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

cy · Answer

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

cy · Answer

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

cy · Answer

In fact you don t need the second loop at all

rocketraman · Answer

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 · Answer

`duplicate` is just the bog standard JDK ByteBuffer method

cy · Answer

This will lead to serious damage

cy · Answer

duplicate doesn t copy buffer content but creates a view

rocketraman · Answer

Oh right it shares content

cy · Answer

You need your own copy function instead

rocketraman · Answer

Ok this is what I have now

rocketraman · Answer

``` 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 · Answer

Seems to work but any comments

cy · Answer

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

rocketraman · Answer

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 · Answer

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

rocketraman · Answer

Ah the `buffer hasRemaining` right

rocketraman · Answer

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

rocketraman · Answer

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

cy · Answer

Yes please

cy · Answer

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

cy · Answer

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

cy · Answer

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

rocketraman · Answer

FYI <https github com Kotlin kotlinx io issues 22>