diff --git a/platform/eel/src/com/intellij/platform/eel/provider/utils/eelChannelUtils.kt b/platform/eel/src/com/intellij/platform/eel/provider/utils/eelChannelUtils.kt index b3f2b926df56..3dd644effda9 100644 --- a/platform/eel/src/com/intellij/platform/eel/provider/utils/eelChannelUtils.kt +++ b/platform/eel/src/com/intellij/platform/eel/provider/utils/eelChannelUtils.kt @@ -79,9 +79,16 @@ fun CoroutineScope.consumeReceiveChannelAsKotlin(receiveChannel: EelReceiveChann /** * Collect data from channel line-by-line using [charset] to convert bytes to chars. - * Much like [java.io.BufferedReader], we consider CR or CRLF as a new line chars. - * This API might be slow (as it reads one byte per time) so you might prefer to [readAllBytes] first, then decode it and split by lines. - * However, for interactive source you can't read till the end, so you use this api. + * A line ends at `\n`, and neither it nor a `\r` in front of it is part of the emitted line, much like + * [java.io.BufferedReader.readLine]. A lone `\r` does not end a line: command line tools use it to + * redraw a line in place, and splitting there would tear that output apart. + * The end of the stream ends the last line, and emits nothing when there is no unterminated line left. + * + * For a non-interactive source you might prefer to [readAllBytes] first, then decode it and split by lines. + * + * This reads ahead, so it fits a channel that is read to its end. To read only the first lines -- a + * handshake, a header -- and leave the rest readable, use [com.intellij.platform.eel.channels.readLine] + * of [com.intellij.platform.eel.channels.PeekableEelReceiveChannel], which puts the remainder back. * * As soon as channel gets closed -- flow finishes. * ```kotlin @@ -99,7 +106,17 @@ fun CoroutineScope.consumeReceiveChannelAsKotlin(receiveChannel: EelReceiveChann */ @ApiStatus.Internal fun EelReceiveChannel.lines(charset: Charset): Flow = - linesImpl(charset) + linesImpl(charset, DEFAULT_BUFFER_SIZE) + +/** + * Please read the documentation for the other overload of [lines]. + * + * [bufferSize] is how many bytes a single `receive` may take, so it bounds the read-ahead described + * there. It does not limit the length of an emitted line. + */ +@ApiStatus.Internal +fun EelReceiveChannel.lines(charset: Charset, bufferSize: Int): Flow = + linesImpl(charset, bufferSize) @ApiStatus.Internal fun EelReceiveChannel.lines(): Flow = lines(Charset.defaultCharset()) diff --git a/platform/eel/src/com/intellij/platform/eel/provider/utils/eelChannelUtilsImpl.kt b/platform/eel/src/com/intellij/platform/eel/provider/utils/eelChannelUtilsImpl.kt index 5f220f633594..73bb062162e1 100644 --- a/platform/eel/src/com/intellij/platform/eel/provider/utils/eelChannelUtilsImpl.kt +++ b/platform/eel/src/com/intellij/platform/eel/provider/utils/eelChannelUtilsImpl.kt @@ -310,26 +310,57 @@ internal fun Socket.consumeAsEelChannelImpl(): EelReceiveChannel = internal fun Socket.asEelChannelImpl(): EelSendChannel = channel?.asEelChannel() ?: outputStream.asEelChannel() -internal fun EelReceiveChannel.linesImpl(charset: Charset): Flow = flow { - val tmpBuffer = ByteBuffer.allocate(1) - var result = ByteArrayOutputStream() - while (true) { - tmpBuffer.rewind() - suspend fun emitBuffer() { - emit(charset.decode(ByteBuffer.wrap(result.toByteArray())).toString()) - } +private const val LINE_FEED = '\n'.code.toByte() +private const val CARRIAGE_RETURN = '\r'.code.toByte() - val r = receive(tmpBuffer) +internal fun EelReceiveChannel.linesImpl(charset: Charset, bufferSize: Int): Flow = flow { + val chunk = ByteBuffer.allocate(bufferSize) + var result = ByteArrayOutputStream() + + // The terminator is dropped by decoding fewer bytes rather than by trimming the decoded line, which + // would allocate a second string for every line that has one. + suspend fun emitBuffer() { + val bytes = result.toByteArray() + var size = bytes.size + if (size > 0 && bytes[size - 1] == LINE_FEED) { + size-- + if (size > 0 && bytes[size - 1] == CARRIAGE_RETURN) { + size-- + } + } + emit(charset.decode(ByteBuffer.wrap(bytes, 0, size)).toString()) + } + + while (true) { + chunk.clear() + val r = receive(chunk) if (r == ReadResult.EOF) { - emitBuffer() + // A stream ending right after a line feed has no unterminated last line to report. + if (result.size() > 0) { + emitBuffer() + } return@flow } - val b = tmpBuffer.flip().get().toInt() - result.write(b) - if (b == 10) { - emitBuffer() - result = ByteArrayOutputStream() + chunk.flip() + while (chunk.hasRemaining()) { + val start = chunk.position() + val limit = chunk.limit() + var end = limit + var lineComplete = false + for (i in start until limit) { + if (chunk.get(i) == LINE_FEED) { + end = i + 1 + lineComplete = true + break + } + } + result.write(chunk.array(), chunk.arrayOffset() + start, end - start) + chunk.position(end) + if (lineComplete) { + emitBuffer() + result = ByteArrayOutputStream() + } } } } diff --git a/platform/platform-tests/testSrc/com/intellij/execution/eel/EelChannelToolsTest.kt b/platform/platform-tests/testSrc/com/intellij/execution/eel/EelChannelToolsTest.kt index dc18946189d2..e4e69ed95e8c 100644 --- a/platform/platform-tests/testSrc/com/intellij/execution/eel/EelChannelToolsTest.kt +++ b/platform/platform-tests/testSrc/com/intellij/execution/eel/EelChannelToolsTest.kt @@ -700,6 +700,53 @@ class EelChannelToolsTest { } Assertions.assertArrayEquals(lines.toTypedArray(), result.toTypedArray(), "Wrong lines collected") } + + /** + * The terminator is not part of the line, and a stream ending on one has no last line to report. + */ + @CartesianTest + fun testLinesFraming(@IntRangeSource(from = 1, to = 4) blockSize: Int): Unit = timeoutRunBlocking { + suspend fun collect(text: String): List { + val channel = ByteArrayInputStreamLimited(text.encodeToByteArray(), blockSize).consumeAsEelChannel() + val result = mutableListOf() + channel.lines(Charsets.UTF_8).collect(result::add) + return result + } + + assertEquals(listOf("one", "two"), collect("one\ntwo\n"), "Terminated stream") + assertEquals(listOf("one", "two"), collect("one\ntwo"), "Unterminated last line") + assertEquals(listOf("a", "", "b"), collect("a\n\nb\n"), "Empty line in the middle") + assertEquals(listOf(), collect(""), "Empty stream") + assertEquals(listOf(""), collect("\n"), "Nothing but a terminator") + assertEquals(listOf("a", "b"), collect("a\r\nb\r\n"), "CRLF, possibly split between reads") + assertEquals(listOf("50%\r100%"), collect("50%\r100%\n"), "A lone CR does not end a line") + assertEquals(listOf("a\r"), collect("a\r"), "A CR is only dropped as the first half of a CRLF") + } + + /** + * Guards against reading byte by byte, which makes the read cost scale with the payload size. + */ + @Test + fun testLinesReadInChunks(): Unit = timeoutRunBlocking { + val bytes = ("a".repeat(512 * 1024) + "\n").encodeToByteArray() + val stream = ByteArrayInputStreamLimited(bytes, bytes.size) + stream.consumeAsEelChannel().lines(Charsets.UTF_8).collect { } + assertTrue(stream.reads.get() < bytes.size / 1024, "One line of ${bytes.size} bytes took ${stream.reads} reads") + } + + /** + * The buffer size bounds what a single read takes from the channel, not the length of a line. + */ + @Test + fun testLinesBufferSize(): Unit = timeoutRunBlocking { + val line = "a".repeat(4096) + val bytes = "$line\n".encodeToByteArray() + val stream = ByteArrayInputStreamLimited(bytes, bytes.size) + val result = mutableListOf() + stream.consumeAsEelChannel().lines(Charsets.UTF_8, 64).collect(result::add) + assertEquals(listOf(line), result, "Wrong lines collected") + assertTrue(stream.reads.get() >= bytes.size / 64, "${bytes.size} bytes in 64-byte reads took ${stream.reads} reads") + } } /** @@ -733,8 +780,13 @@ private class ByteArrayChannelLimited(private val blockSize: Int) : WritableByte */ private class ByteArrayInputStreamLimited(data: ByteArray, private val blockSize: Int) : InputStream() { private val iterator = data.iterator() + + /** One read per `receive`, so this also counts the reads of the channel above the stream. */ + val reads: AtomicInteger = AtomicInteger() + override fun read(): Int = if (iterator.hasNext()) iterator.next().toInt() else -1 override fun read(b: ByteArray, off: Int, len: Int): Int { + reads.incrementAndGet() if (!iterator.hasNext()) return -1 val bytesToWrite = min(len, blockSize) var i = 0