KTOR-7327 Support conversion between byte channel interfaces and kotlinx-io primitives #4594
Conversation
| override val writeBuffer: Sink | ||
| get() { | ||
| if (isClosedForWrite) throw closedCause ?: IOException("Channel is closed for write") | ||
| return origin as? Sink ?: origin.buffered() |
There was a problem hiding this comment.
This should be done eagerly as a single instance returned for every call. If you do not store the results of buffer(), you're liable to lose bytes if you forget to call flush on every instance created.
Consider code that does something as simple as
writeBuffer.writeString("hi")
writeBuffer.apply {
writeString("hey")
flush()
}If the origin is truly a RawSink, it will never see "hi" as those buffered bytes are dropped on the floor.
In general I would recommend always buffering rather than conditionally doing anything. The cost of buffering something that is already buffered is near zero.
Additionally, test using object : RawSink by buffer to ensure nothing can cheat by doing things like conditionally casting to Sink or Buffer.
There was a problem hiding this comment.
good catch, it was caching the result before introducing exception
| * channel.writeString("hello, world!") | ||
| * channel.flushAndClose() | ||
|
|
||
| * (rawSink as Buffer).readString() // "Hello, world" |
There was a problem hiding this comment.
I don't love that we're showing people this is okay.
How about doing a local function like
fun stuff(raw: RawSink) {
val channel = raw.asByteWriteChannel()
...
}
val buffer = Buffer()
stuff(buffer)
buffer.readString() // ...Nothing inherently unsafe here
| @OptIn(InternalAPI::class) | ||
| override suspend fun flushAndClose() { | ||
| if (!closed.compareAndSet(expect = null, update = CLOSED)) return | ||
| writeBuffer.flush() |
There was a problem hiding this comment.
if we close writeBuffer it will release all buffers and we can miss the content
| import kotlinx.io.RawSink | ||
|
|
||
| /** | ||
| * Converts the current `ByteWriteChannel` instance into a `RawSink`. |
There was a problem hiding this comment.
I would warn this implementation uses runBlocking since nesting calls to that function is problematic.
There was a problem hiding this comment.
Yep, will note that in the KDoc. The alternative is a busy loop which is even worse
| * | ||
| * @return a `RawSource` implementation that wraps the `ByteReadChannel`. | ||
| */ | ||
| public fun ByteReadChannel.asSource(): RawSource = ByteReadChannelSource(this) |
There was a problem hiding this comment.
I would warn this implementation uses runBlocking since nesting calls to that function is problematic.
| public fun RawSink.asByteWriteChannel(): ByteWriteChannel = SinkByteWriteChannel(this).apply { | ||
| } | ||
|
|
||
| private suspend fun x() { |
There was a problem hiding this comment.
Delete? Seems to have been for writing docs above.
There was a problem hiding this comment.
thanks for the notice, missed to drop my playground
Co-authored-by: Jake Wharton <[email protected]>
Co-authored-by: Jake Wharton <[email protected]>
There was a problem hiding this comment.
Hey @JakeWharton, thanks for the review! I'll check on the comments
| override val writeBuffer: Sink | ||
| get() { | ||
| if (isClosedForWrite) throw closedCause ?: IOException("Channel is closed for write") | ||
| return origin as? Sink ?: origin.buffered() |
There was a problem hiding this comment.
good catch, it was caching the result before introducing exception
| @OptIn(InternalAPI::class) | ||
| override suspend fun flushAndClose() { | ||
| if (!closed.compareAndSet(expect = null, update = CLOSED)) return | ||
| writeBuffer.flush() |
There was a problem hiding this comment.
if we close writeBuffer it will release all buffers and we can miss the content
| import kotlinx.io.RawSink | ||
|
|
||
| /** | ||
| * Converts the current `ByteWriteChannel` instance into a `RawSink`. |
There was a problem hiding this comment.
Yep, will note that in the KDoc. The alternative is a busy loop which is even worse
Fix KTOR-7327 Support conversion between byte channel interfaces and kotlinx-io primitives