Fix deserialization in read buffer (#225)
* Add deserialization error processing * Add test
This commit is contained in:
parent
ad8205297e
commit
9bb851aba2
2 changed files with 31 additions and 1 deletions
|
|
@ -1,5 +1,6 @@
|
||||||
package io.modelcontextprotocol.kotlin.sdk.shared
|
package io.modelcontextprotocol.kotlin.sdk.shared
|
||||||
|
|
||||||
|
import io.github.oshai.kotlinlogging.KotlinLogging
|
||||||
import io.modelcontextprotocol.kotlin.sdk.JSONRPCMessage
|
import io.modelcontextprotocol.kotlin.sdk.JSONRPCMessage
|
||||||
import kotlinx.io.Buffer
|
import kotlinx.io.Buffer
|
||||||
import kotlinx.io.indexOf
|
import kotlinx.io.indexOf
|
||||||
|
|
@ -9,6 +10,9 @@ import kotlinx.io.readString
|
||||||
* Buffers a continuous stdio stream into discrete JSON-RPC messages.
|
* Buffers a continuous stdio stream into discrete JSON-RPC messages.
|
||||||
*/
|
*/
|
||||||
public class ReadBuffer {
|
public class ReadBuffer {
|
||||||
|
|
||||||
|
private val logger = KotlinLogging.logger { }
|
||||||
|
|
||||||
private val buffer: Buffer = Buffer()
|
private val buffer: Buffer = Buffer()
|
||||||
|
|
||||||
public fun append(chunk: ByteArray) {
|
public fun append(chunk: ByteArray) {
|
||||||
|
|
@ -37,7 +41,13 @@ public class ReadBuffer {
|
||||||
string
|
string
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
return deserializeMessage(line)
|
try {
|
||||||
|
return deserializeMessage(line)
|
||||||
|
} catch (e: Exception) {
|
||||||
|
logger.error(e) { "Failed to deserialize message from line: $line\nSkipping..." }
|
||||||
|
}
|
||||||
|
|
||||||
|
return null
|
||||||
}
|
}
|
||||||
|
|
||||||
public fun clear() {
|
public fun clear() {
|
||||||
|
|
|
||||||
|
|
@ -45,4 +45,24 @@ class StdioClientTransportTest : BaseTransportTest() {
|
||||||
process.waitFor()
|
process.waitFor()
|
||||||
process.destroy()
|
process.destroy()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `should ignore first output messages`() = runTest {
|
||||||
|
val processBuilder = ProcessBuilder("/usr/bin/tee")
|
||||||
|
val process = processBuilder.start()
|
||||||
|
process.outputStream.write("Stdio server started".toByteArray())
|
||||||
|
|
||||||
|
val input = process.inputStream.asSource().buffered()
|
||||||
|
val output = process.outputStream.asSink().buffered()
|
||||||
|
|
||||||
|
val client = StdioClientTransport(
|
||||||
|
input = input,
|
||||||
|
output = output,
|
||||||
|
)
|
||||||
|
|
||||||
|
testClientRead(client)
|
||||||
|
|
||||||
|
process.waitFor()
|
||||||
|
process.destroy()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue