From 9040721730bed4a1095d595f9ad6db9cbbc8584d Mon Sep 17 00:00:00 2001 From: Ludwig Valda Vasquez Date: Thu, 20 Nov 2025 13:27:57 +0100 Subject: [PATCH] refactor [lsp]: abstract over i/o to remove ktor dependency We want to remove ktor dependency from the lsp.protocol module as it conflicts with other uses. The module has a single usage of ktor, its i/o utils types: ByteReadChannel and ByteWriteChannel. These are now replaced by requivalent interfaces. In this patch, ktor dependency is still present and implements the aforementioned interfaces in this same module, but the goal is to move the implementation into module consumers. GitOrigin-RevId: 62681cd01c54b00cab7e17682fe6858db01d4186 --- .../lsp/implementation/LspConnection.kt | 13 ++-- .../com/jetbrains/lsp/implementation/io.kt | 65 +++++++++++++++++++ .../lsp/implementation/protocolFraming.kt | 5 +- .../com/jetbrains/lsp/implementation/tcp.kt | 34 +++++++++- 4 files changed, 105 insertions(+), 12 deletions(-) create mode 100644 fleet/lsp.protocol/srcCommonMain/com/jetbrains/lsp/implementation/io.kt diff --git a/fleet/lsp.protocol/srcCommonMain/com/jetbrains/lsp/implementation/LspConnection.kt b/fleet/lsp.protocol/srcCommonMain/com/jetbrains/lsp/implementation/LspConnection.kt index 60d91599e065..47cc0e11f44c 100644 --- a/fleet/lsp.protocol/srcCommonMain/com/jetbrains/lsp/implementation/LspConnection.kt +++ b/fleet/lsp.protocol/srcCommonMain/com/jetbrains/lsp/implementation/LspConnection.kt @@ -1,14 +1,13 @@ package com.jetbrains.lsp.implementation -import io.ktor.utils.io.ByteReadChannel -import io.ktor.utils.io.ByteWriteChannel -import io.ktor.utils.io.core.Closeable import org.jetbrains.annotations.TestOnly -interface LspConnection : Closeable { - val input: ByteReadChannel - val output: ByteWriteChannel +interface LspConnection { + val input: ByteReader + val output: ByteWriter @TestOnly fun isAlive(): Boolean -} \ No newline at end of file + fun close() +} + diff --git a/fleet/lsp.protocol/srcCommonMain/com/jetbrains/lsp/implementation/io.kt b/fleet/lsp.protocol/srcCommonMain/com/jetbrains/lsp/implementation/io.kt new file mode 100644 index 000000000000..7c1d49a85612 --- /dev/null +++ b/fleet/lsp.protocol/srcCommonMain/com/jetbrains/lsp/implementation/io.kt @@ -0,0 +1,65 @@ +package com.jetbrains.lsp.implementation + +import kotlinx.io.* + +/** Replicates the ByteReadChannel interface from ktor utils with the same semantics. */ +interface ByteReader { + val closedCause: Throwable? + val isClosedForRead: Boolean + val readBuffer: Source + + suspend fun awaitContent(min: Int = 1): Boolean + fun cancel(cause: Throwable?) +} + +/** Replicates the ByteWriteChannel interface from ktor utils with the same semantics. */ +interface ByteWriter { + val isClosedForWrite: Boolean + val closedCause: Throwable? + val writeBuffer: Sink + + suspend fun flush() + suspend fun flushAndClose() + fun cancel(cause: Throwable?) +} + +fun ByteWriter.writeByteArray(array: ByteArray) { + writeBuffer.write(array) +} + +fun ByteReader.cancel() { + cancel(IOException("Channel was cancelled")) +} + +suspend fun ByteReader.readUTF8Line(): String? { + val builder = StringBuilder() + do { + val linefeed = readBuffer.indexOf(0x0A) + if (linefeed != -1L) { + builder.append(readBuffer.readString(linefeed)) + if (builder.isNotEmpty() && builder[builder.length - 1] == '\r') { + builder.deleteAt(builder.length - 1) + } + assert(readBuffer.readByte() == 0x0A.toByte()) + + return builder.toString() + } + + builder.append(readBuffer.readString()) + } + while (awaitContent()) + + // Line terminator was never found before the byte stream was closed. + return null +} + +suspend fun ByteReader.readByteArray(count: Int): ByteArray { + val buffer = Buffer() + while (buffer.size != count.toLong()) { + if (readBuffer.exhausted()) awaitContent() + if (isClosedForRead) break + + readBuffer.readAtMostTo(buffer, count - buffer.size) + } + return buffer.readByteArray() +} diff --git a/fleet/lsp.protocol/srcCommonMain/com/jetbrains/lsp/implementation/protocolFraming.kt b/fleet/lsp.protocol/srcCommonMain/com/jetbrains/lsp/implementation/protocolFraming.kt index 98be77f26317..2d5098259c2b 100644 --- a/fleet/lsp.protocol/srcCommonMain/com/jetbrains/lsp/implementation/protocolFraming.kt +++ b/fleet/lsp.protocol/srcCommonMain/com/jetbrains/lsp/implementation/protocolFraming.kt @@ -3,7 +3,6 @@ package com.jetbrains.lsp.implementation import com.jetbrains.lsp.protocol.LSP import fleet.util.decodeToStringUtf8 import fleet.util.encodeToByteArrayUtf8 -import io.ktor.utils.io.* import kotlinx.coroutines.CompletableDeferred import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.channels.Channel @@ -61,7 +60,7 @@ suspend fun withBaseProtocolFraming( } } -private suspend fun ByteReadChannel.readFrame(): JsonElement? { +private suspend fun ByteReader.readFrame(): JsonElement? { var contentLength = -1 var readSomething = false val buf = try { @@ -89,7 +88,7 @@ private suspend fun ByteReadChannel.readFrame(): JsonElement? { /** * @return Boolean indicating whether the frame was successfully written (`true`) or the channel was closed (`false`). */ -private suspend fun ByteWriteChannel.writeFrame(jsonElement: JsonElement): Boolean { +private suspend fun ByteWriter.writeFrame(jsonElement: JsonElement): Boolean { val str = LSP.json.encodeToString(JsonElement.serializer(), jsonElement) val frameStr = buildString { // protocol requires string length in bytes diff --git a/fleet/lsp.protocol/srcJvmMain/com/jetbrains/lsp/implementation/tcp.kt b/fleet/lsp.protocol/srcJvmMain/com/jetbrains/lsp/implementation/tcp.kt index 7d95a89e295a..d2bde7e2c1e9 100644 --- a/fleet/lsp.protocol/srcJvmMain/com/jetbrains/lsp/implementation/tcp.kt +++ b/fleet/lsp.protocol/srcJvmMain/com/jetbrains/lsp/implementation/tcp.kt @@ -5,7 +5,10 @@ import io.ktor.network.selector.SelectorManager import io.ktor.network.sockets.* import io.ktor.utils.io.ByteReadChannel import io.ktor.utils.io.ByteWriteChannel +import io.ktor.utils.io.InternalAPI import kotlinx.coroutines.* +import kotlinx.io.Sink +import kotlinx.io.Source import kotlin.time.Duration import kotlin.time.Duration.Companion.seconds @@ -98,9 +101,36 @@ sealed interface TcpConnectionConfig { ) : TcpConnectionConfig } +@OptIn(InternalAPI::class) +class KtorByteReader(val input: ByteReadChannel) : ByteReader { + override val closedCause: Throwable? + get() = input.closedCause + override val isClosedForRead: Boolean + get() = input.isClosedForRead + override val readBuffer: Source + get() = input.readBuffer + + override suspend fun awaitContent(min: Int): Boolean = input.awaitContent() + override fun cancel(cause: Throwable?): Unit = input.cancel(cause) +} + +@OptIn(InternalAPI::class) +class KtorByteWriter(val output: ByteWriteChannel) : ByteWriter { + override val isClosedForWrite: Boolean + get() = output.isClosedForWrite + override val closedCause: Throwable? + get() = output.closedCause + override val writeBuffer: Sink + get() = output.writeBuffer + + override suspend fun flush(): Unit = output.flush() + override suspend fun flushAndClose(): Unit = output.flushAndClose() + override fun cancel(cause: Throwable?): Unit = output.cancel(cause) +} + class KtorSocketConnection(private val socket: Socket) : LspConnection { - override val input: ByteReadChannel = socket.openReadChannel() - override val output: ByteWriteChannel = socket.openWriteChannel(autoFlush = true) + override val input: ByteReader = KtorByteReader(socket.openReadChannel()) + override val output: ByteWriter = KtorByteWriter(socket.openWriteChannel(autoFlush = true)) override fun close() { socket.close()