From 212cbfa325a32994d534ced818870da80c0b5a2b Mon Sep 17 00:00:00 2001 From: Konstantin Pavlov <1517853+kpavlov@users.noreply.github.com> Date: Mon, 16 Feb 2026 14:10:56 +0200 Subject: [PATCH] feat: Add kotlin-sdk-testing module with channel-based transport implementation (#505) # feat: Add kotlin-sdk-testing module with channel-based transport implementation - Introduced `kotlin-sdk-testing` module for testing utilities in the MCP Kotlin SDK. - Implemented `ChannelTransport`, providing JSON-RPC transport via Kotlin coroutines channels. ## Motivation and Context This ChannelTransport can be used for testing Server implementation without using real transports. ## How Has This Been Tested? Unit test ## Breaking Changes No ## Types of changes - [ ] Bug fix (non-breaking change which fixes an issue) - [x] New feature (non-breaking change which adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [x] Documentation update ## Checklist - [x] I have read the [MCP Documentation](https://modelcontextprotocol.io) - [x] My code follows the repository's style guidelines - [x] New and existing tests pass locally - [x] I have added appropriate error handling - [x] I have added or updated documentation as needed --- README.md | 7 + build.gradle.kts | 4 +- integration-test/build.gradle.kts | 1 + .../sdk/testing/ChannelTransportTest.kt | 88 ++++++++ kotlin-sdk-client/build.gradle.kts | 1 + .../AbstractClientTransportLifecycleTest.kt | 4 +- .../ChannelClientTransportLifecycleTest.kt | 20 ++ kotlin-sdk-testing/Module.md | 67 ++++++ kotlin-sdk-testing/api/kotlin-sdk-testing.api | 26 +++ kotlin-sdk-testing/build.gradle.kts | 53 +++++ .../kotlin/sdk/testing/ChannelTransport.kt | 176 +++++++++++++++ .../sdk/testing/ChannelTransportTest.kt | 212 ++++++++++++++++++ .../resources/junit-platform.properties | 5 + .../jvmTest/resources/simplelogger.properties | 7 + settings.gradle.kts | 1 + 15 files changed, 669 insertions(+), 3 deletions(-) create mode 100644 integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransportTest.kt create mode 100644 kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/channel/ChannelClientTransportLifecycleTest.kt create mode 100644 kotlin-sdk-testing/Module.md create mode 100644 kotlin-sdk-testing/api/kotlin-sdk-testing.api create mode 100644 kotlin-sdk-testing/build.gradle.kts create mode 100644 kotlin-sdk-testing/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport.kt create mode 100644 kotlin-sdk-testing/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransportTest.kt create mode 100644 kotlin-sdk-testing/src/jvmTest/resources/junit-platform.properties create mode 100644 kotlin-sdk-testing/src/jvmTest/resources/simplelogger.properties diff --git a/README.md b/README.md index 881ee98..7df2be2 100644 --- a/README.md +++ b/README.md @@ -46,6 +46,7 @@ standardized protocol interface. * [Streamable HTTP Transport](#streamable-http-transport) * [SSE Transport](#sse-transport) * [WebSocket Transport](#websocket-transport) + * [ChannelTransport (testing)](#channeltransport-testing) * [Connecting your server](#connecting-your-server) * [Examples](#examples) * [Documentation](#documentation) @@ -774,6 +775,12 @@ SSE Ktor plugin when you need drop-in compatibility, but prefer Streamable HTTP `WebSocketClientTransport` plus the matching server utilities provide full-duplex, low-latency connections—useful when you expect lots of notifications or long-running sessions behind a reverse proxy that already terminates WebSockets. +### ChannelTransport (testing) + +`ChannelTransport` provides a simple, non-networked transport for testing and local development. +It uses Kotlin coroutines channels to provide a full-duplex connection between a client and server, +allowing for easy testing of MCP functionality without the need for network setup. + ## Connecting your server 1. Start a sample HTTP server on port 3000: diff --git a/build.gradle.kts b/build.gradle.kts index 57dd810..09e9adb 100644 --- a/build.gradle.kts +++ b/build.gradle.kts @@ -11,11 +11,13 @@ dependencies { dokka(project(":kotlin-sdk-core")) dokka(project(":kotlin-sdk-client")) dokka(project(":kotlin-sdk-server")) + dokka(project(":kotlin-sdk-testing")) kover(project(":kotlin-sdk-core")) kover(project(":kotlin-sdk-client")) kover(project(":kotlin-sdk-server")) - "kover"(project(":integration-test")) + kover(project(":kotlin-sdk-testing")) + kover(project(":integration-test")) } subprojects { diff --git a/integration-test/build.gradle.kts b/integration-test/build.gradle.kts index 5352234..82078e7 100644 --- a/integration-test/build.gradle.kts +++ b/integration-test/build.gradle.kts @@ -30,6 +30,7 @@ kotlin { jvmTest { dependencies { implementation(project(":test-utils")) + implementation(project(":kotlin-sdk-testing")) } } } diff --git a/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransportTest.kt b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransportTest.kt new file mode 100644 index 0000000..ea246db --- /dev/null +++ b/integration-test/src/jvmTest/kotlin/io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransportTest.kt @@ -0,0 +1,88 @@ +package io.modelcontextprotocol.kotlin.sdk.testing + +import io.kotest.matchers.collections.shouldHaveSize +import io.kotest.matchers.nulls.shouldNotBeNull +import io.modelcontextprotocol.kotlin.sdk.ExperimentalMcpApi +import io.modelcontextprotocol.kotlin.sdk.client.Client +import io.modelcontextprotocol.kotlin.sdk.server.Server +import io.modelcontextprotocol.kotlin.sdk.server.ServerOptions +import io.modelcontextprotocol.kotlin.sdk.server.ServerSession +import io.modelcontextprotocol.kotlin.sdk.types.Implementation +import io.modelcontextprotocol.kotlin.sdk.types.InitializeRequest +import io.modelcontextprotocol.kotlin.sdk.types.InitializeResult +import io.modelcontextprotocol.kotlin.sdk.types.LATEST_PROTOCOL_VERSION +import io.modelcontextprotocol.kotlin.sdk.types.ListResourcesRequest +import io.modelcontextprotocol.kotlin.sdk.types.ListResourcesResult +import io.modelcontextprotocol.kotlin.sdk.types.Method +import io.modelcontextprotocol.kotlin.sdk.types.Resource +import io.modelcontextprotocol.kotlin.sdk.types.ServerCapabilities +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.joinAll +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlin.test.Test + +@OptIn(ExperimentalMcpApi::class) +class ChannelTransportTest { + + @Test + fun `should connect and list resources`(): Unit = runBlocking { + val serverOptions = ServerOptions( + capabilities = ServerCapabilities( + resources = ServerCapabilities.Resources(), + ), + ) + val server = Server( + Implementation(name = "test server", version = "1.0"), + serverOptions, + ) + + val (clientTransport, serverTransport) = ChannelTransport.createLinkedPair() + + val client = Client( + clientInfo = Implementation(name = "test client", version = "1.0"), + ) + + val serverSessionResult = CompletableDeferred() + + listOf( + launch { + client.connect(clientTransport) + }, + launch { + serverSessionResult.complete(server.createSession(serverTransport)) + }, + ).joinAll() + + val serverSession = serverSessionResult.await() + serverSession.setRequestHandler(Method.Defined.Initialize) { _, _ -> + InitializeResult( + protocolVersion = LATEST_PROTOCOL_VERSION, + capabilities = ServerCapabilities( + resources = ServerCapabilities.Resources(null, null), + tools = ServerCapabilities.Tools(null), + ), + serverInfo = Implementation(name = "test", version = "1.0"), + ) + } + + serverSession.setRequestHandler(Method.Defined.ResourcesList) { _, _ -> + ListResourcesResult( + resources = listOf( + Resource( + uri = "/foo/bar", + name = "foo-bar-resource", + ), + ), + ) + } + + // These should not throw + client.listResources() shouldNotBeNull { + this.resources shouldHaveSize 1 + } + + client.close() + server.close() + } +} diff --git a/kotlin-sdk-client/build.gradle.kts b/kotlin-sdk-client/build.gradle.kts index 36a3ed8..f50266c 100644 --- a/kotlin-sdk-client/build.gradle.kts +++ b/kotlin-sdk-client/build.gradle.kts @@ -41,6 +41,7 @@ kotlin { commonTest { dependencies { implementation(kotlin("test")) + implementation(project(":kotlin-sdk-testing")) implementation(libs.kotest.assertions.core) implementation(libs.kotlinx.coroutines.test) implementation(libs.ktor.client.logging) diff --git a/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/AbstractClientTransportLifecycleTest.kt b/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/AbstractClientTransportLifecycleTest.kt index b6ece42..28a1337 100644 --- a/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/AbstractClientTransportLifecycleTest.kt +++ b/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/AbstractClientTransportLifecycleTest.kt @@ -4,7 +4,7 @@ import io.kotest.assertions.nondeterministic.eventually import io.kotest.assertions.throwables.shouldThrow import io.kotest.matchers.shouldBe import io.kotest.matchers.string.shouldContain -import io.modelcontextprotocol.kotlin.sdk.shared.AbstractTransport +import io.modelcontextprotocol.kotlin.sdk.shared.AbstractClientTransport import io.modelcontextprotocol.kotlin.sdk.types.McpException import io.modelcontextprotocol.kotlin.sdk.types.PingRequest import io.modelcontextprotocol.kotlin.sdk.types.RPCError.ErrorCode.CONNECTION_CLOSED @@ -16,7 +16,7 @@ import kotlin.test.Test import kotlin.time.Duration.Companion.milliseconds import kotlin.time.Duration.Companion.seconds -abstract class AbstractClientTransportLifecycleTest { +abstract class AbstractClientTransportLifecycleTest { protected lateinit var transport: T diff --git a/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/channel/ChannelClientTransportLifecycleTest.kt b/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/channel/ChannelClientTransportLifecycleTest.kt new file mode 100644 index 0000000..853a433 --- /dev/null +++ b/kotlin-sdk-client/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/client/channel/ChannelClientTransportLifecycleTest.kt @@ -0,0 +1,20 @@ +package io.modelcontextprotocol.kotlin.sdk.client.channel + +import io.modelcontextprotocol.kotlin.sdk.ExperimentalMcpApi +import io.modelcontextprotocol.kotlin.sdk.client.AbstractClientTransportLifecycleTest +import io.modelcontextprotocol.kotlin.sdk.testing.ChannelTransport +import kotlin.test.Ignore +import kotlin.test.Test + +@OptIn(ExperimentalMcpApi::class) +class ChannelClientTransportLifecycleTest : AbstractClientTransportLifecycleTest() { + + /** + * Dummy method to make IDE treat this class as a test + */ + @Test + @Ignore + fun dummyTest() = Unit + + override fun createTransport(): ChannelTransport = ChannelTransport() +} diff --git a/kotlin-sdk-testing/Module.md b/kotlin-sdk-testing/Module.md new file mode 100644 index 0000000..9291de5 --- /dev/null +++ b/kotlin-sdk-testing/Module.md @@ -0,0 +1,67 @@ +# Module kotlin-sdk-testing + +`kotlin-sdk-testing` provides testing utilities for MCP client-server interactions. It eliminates the need for external +processes or network connections by offering coroutine-safe in-memory transport implementations that connect clients and servers directly. + +## Typical usage + +1. Create linked transport pair with `ChannelTransport.createLinkedPair()`. +2. Connect client and server concurrently using the paired transports. +3. Configure server request handlers. +4. Execute client operations and verify results. + +### Complete test example + +```kotlin +@OptIn(ExperimentalMcpApi::class) +@Test +fun testClientServerCommunication(): Unit = runBlocking { + val server = Server( + Implementation(name = "test server", version = "1.0"), + ServerOptions(capabilities = ServerCapabilities(resources = ServerCapabilities.Resources())), + ) + + val (clientTransport, serverTransport) = ChannelTransport.createLinkedPair() + + val client = Client(clientInfo = Implementation(name = "test client", version = "1.0")) + + val serverSessionResult = CompletableDeferred() + listOf( + launch { client.connect(clientTransport) }, + launch { serverSessionResult.complete(server.createSession(serverTransport)) }, + ).joinAll() + + val serverSession = serverSessionResult.await() + serverSession.setRequestHandler(Method.Defined.Initialize) { _, _ -> + InitializeResult( + protocolVersion = LATEST_PROTOCOL_VERSION, + capabilities = ServerCapabilities( + resources = ServerCapabilities.Resources(null, null), + tools = ServerCapabilities.Tools(null), + ), + serverInfo = Implementation(name = "test", version = "1.0"), + ) + } + + serverSession.setRequestHandler(Method.Defined.ResourcesList) { _, _ -> + ListResourcesResult( + resources = listOf(Resource(uri = "/foo/bar", name = "foo-bar-resource")), + ) + } + + // Test client operations + client.listResources() shouldNotBeNull { + this.resources shouldHaveSize 1 + } + + // Clean up resources + client.close() + server.close() +} +``` + +**Note**: In production tests, wrap client/server operations in try-finally blocks to ensure proper resource cleanup even if assertions fail. + +Use this module for unit and integration testing of MCP implementations without managing external server processes, +network ports, or I/O streams. The channel-based transport ensures tests run instantly with full isolation between test +cases. diff --git a/kotlin-sdk-testing/api/kotlin-sdk-testing.api b/kotlin-sdk-testing/api/kotlin-sdk-testing.api new file mode 100644 index 0000000..4a942a4 --- /dev/null +++ b/kotlin-sdk-testing/api/kotlin-sdk-testing.api @@ -0,0 +1,26 @@ +public final class io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport : io/modelcontextprotocol/kotlin/sdk/shared/AbstractClientTransport { + public static final field Companion Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport$Companion; + public fun (Lkotlinx/coroutines/channels/Channel;Lkotlinx/coroutines/CoroutineDispatcher;)V + public synthetic fun (Lkotlinx/coroutines/channels/Channel;Lkotlinx/coroutines/CoroutineDispatcher;ILkotlin/jvm/internal/DefaultConstructorMarker;)V + public fun (Lkotlinx/coroutines/channels/SendChannel;Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlinx/coroutines/CoroutineDispatcher;)V + public synthetic fun (Lkotlinx/coroutines/channels/SendChannel;Lkotlinx/coroutines/channels/ReceiveChannel;Lkotlinx/coroutines/CoroutineDispatcher;ILkotlin/jvm/internal/DefaultConstructorMarker;)V +} + +public final class io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport$Companion { + public final fun createLinkedPair (ILkotlinx/coroutines/CoroutineDispatcher;)Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport$LinkedTransports; + public static synthetic fun createLinkedPair$default (Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport$Companion;ILkotlinx/coroutines/CoroutineDispatcher;ILjava/lang/Object;)Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport$LinkedTransports; +} + +public final class io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport$LinkedTransports { + public fun (Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport;Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport;)V + public final fun component1 ()Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport; + public final fun component2 ()Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport; + public final fun copy (Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport;Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport;)Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport$LinkedTransports; + public static synthetic fun copy$default (Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport$LinkedTransports;Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport;Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport;ILjava/lang/Object;)Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport$LinkedTransports; + public fun equals (Ljava/lang/Object;)Z + public final fun getClientTransport ()Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport; + public final fun getServerTransport ()Lio/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport; + public fun hashCode ()I + public fun toString ()Ljava/lang/String; +} + diff --git a/kotlin-sdk-testing/build.gradle.kts b/kotlin-sdk-testing/build.gradle.kts new file mode 100644 index 0000000..902665c --- /dev/null +++ b/kotlin-sdk-testing/build.gradle.kts @@ -0,0 +1,53 @@ +@file:OptIn(ExperimentalWasmDsl::class) + +import org.jetbrains.kotlin.gradle.ExperimentalWasmDsl + +plugins { + id("mcp.multiplatform") + id("mcp.publishing") + id("mcp.dokka") + alias(libs.plugins.kotlinx.binary.compatibility.validator) +} + +kotlin { + + iosArm64() + iosX64() + iosSimulatorArm64() + watchosX64() + watchosArm64() + watchosSimulatorArm64() + tvosX64() + tvosArm64() + tvosSimulatorArm64() + js { + browser() + nodejs() + } + wasmJs { + browser() + nodejs() + } + + sourceSets { + commonMain { + dependencies { + api(project(":kotlin-sdk-core")) + implementation(libs.kotlin.logging) + } + } + + commonTest { + dependencies { + implementation(kotlin("test")) + implementation(libs.kotlinx.coroutines.test) + implementation(libs.kotest.assertions.core) + } + } + jvmTest { + dependencies { + runtimeOnly(libs.slf4j.simple) + } + } + } +} diff --git a/kotlin-sdk-testing/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport.kt b/kotlin-sdk-testing/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport.kt new file mode 100644 index 0000000..e6d19f7 --- /dev/null +++ b/kotlin-sdk-testing/src/commonMain/kotlin/io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransport.kt @@ -0,0 +1,176 @@ +package io.modelcontextprotocol.kotlin.sdk.testing + +import io.github.oshai.kotlinlogging.KLogger +import io.github.oshai.kotlinlogging.KotlinLogging +import io.modelcontextprotocol.kotlin.sdk.ExperimentalMcpApi +import io.modelcontextprotocol.kotlin.sdk.shared.AbstractClientTransport +import io.modelcontextprotocol.kotlin.sdk.shared.TransportSendOptions +import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCMessage +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.CoroutineName +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.channels.Channel.Factory.UNLIMITED +import kotlinx.coroutines.channels.ReceiveChannel +import kotlinx.coroutines.channels.SendChannel +import kotlinx.coroutines.launch +import kotlinx.coroutines.yield + +/** + * A transport implementation that uses Kotlin Coroutines Channels for asynchronous + * communication of JSON-RPC messages. This is useful in scenarios where communication is + * message-based and channels are used as the underlying mechanism for delivery. + * + * @constructor Initializes the transport with the specified [SendChannel] and [ReceiveChannel]. + * @property sendChannel The channel used to send JSON-RPC messages. + * @property receiveChannel The channel used to receive JSON-RPC messages. + * @param dispatcher The coroutine dispatcher for the event loop. Defaults to [Dispatchers.Default]. + */ +@ExperimentalMcpApi +public class ChannelTransport( + private val sendChannel: SendChannel, + private val receiveChannel: ReceiveChannel, + dispatcher: CoroutineDispatcher = Dispatchers.Default, +) : AbstractClientTransport() { + + override val logger: KLogger = KotlinLogging.logger {} + + private val scope = CoroutineScope(SupervisorJob() + dispatcher) + + /** + * Creates a `ChannelTransport` instance using a single channel for both sending and receiving messages. + * + * This constructor simplifies the use of `ChannelTransport` in scenarios where a single channel + * is enough for bidirectional communication. A default channel with unlimited capacity is used + * if no channel is provided. + * + * @param channel The `Channel` of `JSONRPCMessage` instances, used for both sending and receiving. + * Defaults to a channel with unlimited capacity. + * @param dispatcher The coroutine dispatcher for the event loop. Defaults to [Dispatchers.Default]. + */ + public constructor( + channel: Channel = Channel(UNLIMITED), + dispatcher: CoroutineDispatcher = Dispatchers.Default, + ) : this(channel, channel, dispatcher) + + /** + * Represents a pair of interconnected [ChannelTransport]s for bidirectional communication. + * + * Messages sent via one transport are received by the other. This is useful for + * setting up communication between a client and server without requiring external networking. + * + * @property clientTransport The transport intended for use on the client-side. + * @property serverTransport The transport intended for use on the server-side. + */ + public data class LinkedTransports(val clientTransport: ChannelTransport, val serverTransport: ChannelTransport) + + public companion object { + + /** + * Creates a pair of interconnected [ChannelTransport] objects for bidirectional communication. + * + * This method sets up a client and server transport using Kotlin channels, enabling messages sent + * through one transport to be received by the other. It is typically used for testing or in-memory + * communication without external networking. + * + * @param capacity The buffer capacity of the internal channels used for communication. Defaults to 256. + * @return A [LinkedTransports] instance containing the client and server transports. + */ + public fun createLinkedPair( + capacity: Int = 256, + dispatcher: CoroutineDispatcher = Dispatchers.Default, + ): LinkedTransports { + val sendChannel = Channel(capacity) + val receiveChannel = Channel(capacity) + val clientTransport = ChannelTransport(sendChannel, receiveChannel, dispatcher) + val serverTransport = ChannelTransport(receiveChannel, sendChannel, dispatcher) + return LinkedTransports(clientTransport = clientTransport, serverTransport = serverTransport) + } + } + + /** + * Starts the transport, allowing it to begin processing messages. + * + * Waits for the event loop to start before returning, ensuring the transport is + * fully initialized and ready to send/receive messages. + * + * Processes messages until the channel closes or cancellation occurs. + * Exceptions from message handlers are reported via onError but do not stop processing. + */ + override suspend fun initialize() { + logger.info { "ChannelTransport starting message processing" } + val started = CompletableDeferred() + @Suppress("TooGenericExceptionCaught") + scope.launch(CoroutineName("ChannelTransport#${hashCode()}-event-loop")) { + try { + // Signal that event loop has started, then yield to ensure we're ready + started.complete(Unit) + yield() + + for (message in receiveChannel) { + logger.debug { "Received message: ${message::class.simpleName}" } + + try { + _onMessage.invoke(message) + logger.trace { "Message processed successfully: ${message::class.simpleName}" } + } catch (e: CancellationException) { + // Let cancellation propagate immediately + logger.debug { "Cancellation requested during message processing" } + throw e + } catch (e: Exception) { + // Report other errors but continue processing + logger.warn(e) { "Error processing message: ${message::class.simpleName}" } + _onError.invoke(e) + } + } + logger.info { "ChannelTransport stopped: receive channel closed" } + } catch (e: Exception) { + // Only complete exceptionally if not already completed + if (!started.isCompleted) { + started.completeExceptionally(e) + } + throw e + } finally { + close() + } + } + // Wait for the event loop to start + started.await() + } + + /** + * Sends a JSON-RPC message through the transport. + * + * @param message The JSON-RPC message to send. This can be a request or a response, + * conforming to the JSON-RPC 2.0 specification. + * @param options Optional transport-specific options that provide additional context or behavior + * for the message transmission, such as associating the message with a specific + * incoming request or handling resumption tokens. + */ + override suspend fun performSend(message: JSONRPCMessage, options: TransportSendOptions?) { + logger.debug { "Sending message: ${message::class.simpleName}" } + sendChannel.send(message) + logger.trace { "Message sent successfully: ${message::class.simpleName}" } + } + + /** + * Closes the transport, preventing further message transmission and processing. + */ + override suspend fun closeResources() { + logger.info { "Closing ChannelTransport" } + sendChannel.close() + if (receiveChannel !== sendChannel) { + logger.debug { "Cancelling separate receive channel" } + receiveChannel.cancel() + } + scope.cancel() + scope.coroutineContext[Job]?.join() // Wait for cleanup + logger.info { "ChannelTransport closed" } + } +} diff --git a/kotlin-sdk-testing/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransportTest.kt b/kotlin-sdk-testing/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransportTest.kt new file mode 100644 index 0000000..7fc41bb --- /dev/null +++ b/kotlin-sdk-testing/src/commonTest/kotlin/io/modelcontextprotocol/kotlin/sdk/testing/ChannelTransportTest.kt @@ -0,0 +1,212 @@ +package io.modelcontextprotocol.kotlin.sdk.testing + +import io.kotest.assertions.nondeterministic.eventually +import io.kotest.matchers.collections.shouldContainExactly +import io.kotest.matchers.shouldBe +import io.kotest.matchers.types.shouldBeInstanceOf +import io.modelcontextprotocol.kotlin.sdk.ExperimentalMcpApi +import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCMessage +import io.modelcontextprotocol.kotlin.sdk.types.JSONRPCRequest +import io.modelcontextprotocol.kotlin.sdk.types.RequestId +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.channels.ClosedSendChannelException +import kotlinx.coroutines.launch +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.time.Duration.Companion.seconds + +@OptIn(ExperimentalMcpApi::class) +class ChannelTransportTest { + + @Test + fun `send writes to sendChannel`() = runTest { + val sendChannel = Channel(Channel.UNLIMITED) + val transport = ChannelTransport(sendChannel, Channel()) + + transport.start() + + val message = JSONRPCRequest(RequestId.NumberId(42), "test") + transport.send(message) + + sendChannel.receive() shouldBe message + } + + @Test + fun `start reads from receiveChannel and invokes onMessage`() = runTest { + val receiveChannel = Channel(Channel.UNLIMITED) + val transport = ChannelTransport(Channel(), receiveChannel) + + val messagesProcessed = CompletableDeferred() + val received = mutableListOf() + transport.onMessage { msg -> + received.add(msg) + if (received.size == 2) { + messagesProcessed.complete(Unit) + } + } + + transport.start() + + val msg1 = JSONRPCRequest(RequestId.NumberId(1), "method1") + val msg2 = JSONRPCRequest(RequestId.NumberId(2), "method2") + receiveChannel.send(msg1) + receiveChannel.send(msg2) + + // Wait for messages to be processed + messagesProcessed.await() + + received.shouldContainExactly(msg1, msg2) + } + + @Test + fun `start completes when channel closes`() = runTest { + val receiveChannel = Channel(Channel.UNLIMITED) + val transport = ChannelTransport(Channel(), receiveChannel) + + val messageProcessed = CompletableDeferred() + val received = mutableListOf() + transport.onMessage { + received.add(it) + messageProcessed.complete(Unit) + } + + transport.start() + + val msg = JSONRPCRequest(RequestId.NumberId(1), "method1") + receiveChannel.send(msg) + receiveChannel.close() // Close the channel + + // Wait for a message to be processed + messageProcessed.await() + + received.shouldContainExactly(msg) + } + + @Test + fun `close is idempotent`() = runTest { + val transport = ChannelTransport() + + var closeCount = 0 + transport.onClose { closeCount++ } + + transport.start() + + transport.close() + transport.close() + + closeCount shouldBe 1 + } + + @Test + fun `callbacks chain not replace`() = runTest { + val receiveChannel = Channel(Channel.UNLIMITED) + val transport = ChannelTransport(Channel(), receiveChannel) + + val messageProcessed = CompletableDeferred() + val calls = mutableListOf() + transport.onMessage { + calls.add("first") + } + transport.onMessage { + calls.add("second") + messageProcessed.complete(Unit) + } + + val startJob = backgroundScope.launch { transport.start() } + + receiveChannel.send(JSONRPCRequest(RequestId.NumberId(1), "test")) + messageProcessed.await() + + calls.shouldContainExactly("first", "second") + startJob.cancelAndJoin() + } + + @Test + fun `secondary constructor uses single channel`() = runTest { + val channel = Channel(Channel.UNLIMITED) + val transport = ChannelTransport(channel) + + val messageProcessed = CompletableDeferred() + val received = mutableListOf() + transport.onMessage { msg -> + received.add(msg) + messageProcessed.complete(Unit) + } + + transport.start() + + val message = JSONRPCRequest(RequestId.NumberId(1), "test") + transport.send(message) + messageProcessed.await() + + eventually(2.seconds) { + received.shouldContainExactly(message) + } + } + + @Test + fun `send to closed channel triggers error`() = runTest { + val sendChannel = Channel(Channel.UNLIMITED) + val transport = ChannelTransport(sendChannel, Channel()) + + transport.start() + + var errorCaught = false + transport.onError { + it.shouldBeInstanceOf() + errorCaught = true + } + sendChannel.close() + + try { + transport.send(JSONRPCRequest(RequestId.NumberId(1), "method")) + } catch (e: Exception) { + // send() wraps ClosedSendChannelException in McpException + e.shouldBeInstanceOf() + } + + eventually(2.seconds) { + errorCaught shouldBe true + } + } + + @Test + fun `exceptions in message handler do not stop processing`() = runTest { + val receiveChannel = Channel(Channel.UNLIMITED) + val transport = ChannelTransport(Channel(), receiveChannel) + + val allProcessed = CompletableDeferred() + val received = mutableListOf() + val errors = mutableListOf() + + transport.onError { errors.add(it) } + transport.onMessage { msg -> + val id = ((msg as JSONRPCRequest).id as RequestId.NumberId).value.toInt() + received.add(id) + if (id == 2) { + @Suppress("TooGenericExceptionThrown") + throw RuntimeException("Error processing message 2") + } + if (received.size == 4) { + allProcessed.complete(Unit) + } + } + + transport.start() + + // Send 4 messages, second one will throw + receiveChannel.send(JSONRPCRequest(RequestId.NumberId(1), "m1")) + receiveChannel.send(JSONRPCRequest(RequestId.NumberId(2), "m2")) + receiveChannel.send(JSONRPCRequest(RequestId.NumberId(3), "m3")) + receiveChannel.send(JSONRPCRequest(RequestId.NumberId(4), "m4")) + + allProcessed.await() + + // All messages should be processed despite error in message 2 + received.shouldContainExactly(1, 2, 3, 4) + errors.size shouldBe 1 + errors[0].message shouldBe "Error processing message 2" + } +} diff --git a/kotlin-sdk-testing/src/jvmTest/resources/junit-platform.properties b/kotlin-sdk-testing/src/jvmTest/resources/junit-platform.properties new file mode 100644 index 0000000..f14d100 --- /dev/null +++ b/kotlin-sdk-testing/src/jvmTest/resources/junit-platform.properties @@ -0,0 +1,5 @@ +## https://docs.junit.org/5.3.0-M1/user-guide/index.html#writing-tests-parallel-execution +junit.jupiter.execution.parallel.enabled=true +junit.jupiter.execution.parallel.config.strategy=dynamic +junit.jupiter.execution.parallel.mode.default=concurrent +junit.jupiter.execution.parallel.mode.classes.default=concurrent diff --git a/kotlin-sdk-testing/src/jvmTest/resources/simplelogger.properties b/kotlin-sdk-testing/src/jvmTest/resources/simplelogger.properties new file mode 100644 index 0000000..5679351 --- /dev/null +++ b/kotlin-sdk-testing/src/jvmTest/resources/simplelogger.properties @@ -0,0 +1,7 @@ +# Level of logging for the ROOT logger: ERROR, WARN, INFO, DEBUG, TRACE (default is INFO) +org.slf4j.simpleLogger.defaultLogLevel=INFO +org.slf4j.simpleLogger.showThreadName=true +org.slf4j.simpleLogger.showDateTime=false + +# Log level for specific packages or classes +org.slf4j.simpleLogger.log.io.modelcontextprotocol=DEBUG diff --git a/settings.gradle.kts b/settings.gradle.kts index 95d80b1..60e3d1c 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -22,6 +22,7 @@ include( ":kotlin-sdk-core", ":kotlin-sdk-client", ":kotlin-sdk-server", + ":kotlin-sdk-testing", ":kotlin-sdk", ":integration-test", ":docs",