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

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
-}
+  fun close()
+}
+

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()
+}

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

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()